# SPDX-License-Identifier: Apache-2.0

import datetime
import uuid

import pretend
import pytest

from freezegun import freeze_time
from paginate_sqlalchemy import SqlalchemyOrmPage as SQLAlchemyORMPage
from psycopg.errors import UniqueViolation
from pyramid.httpexceptions import HTTPBadRequest, HTTPNotFound, HTTPSeeOther
from webob.multidict import MultiDict

from tests.common.db.accounts import EmailFactory, UserFactory
from tests.common.db.oidc import PendingGitHubPublisherFactory
from tests.common.db.organizations import (
    OrganizationApplicationFactory,
    OrganizationApplicationObservationFactory,
    OrganizationEventFactory,
    OrganizationFactory,
    OrganizationInvitationFactory,
    OrganizationProjectFactory,
    OrganizationRoleFactory,
    OrganizationStripeCustomerFactory,
    OrganizationStripeSubscriptionFactory,
    TeamFactory,
)
from tests.common.db.packaging import ProjectFactory, RoleFactory
from tests.common.db.subscriptions import (
    StripeCustomerFactory,
    StripeSubscriptionFactory,
    StripeSubscriptionPriceFactory,
)
from warehouse.accounts import ITokenService, IUserService
from warehouse.accounts.interfaces import TokenExpired
from warehouse.admin.flags import AdminFlagValue
from warehouse.authnz import Permissions
from warehouse.manage import views
from warehouse.manage.views import organizations as org_views
from warehouse.oidc.models import PendingGitHubPublisher
from warehouse.organizations import IOrganizationService
from warehouse.organizations.models import (
    Organization,
    OrganizationApplicationStatus,
    OrganizationInvitation,
    OrganizationInvitationStatus,
    OrganizationRole,
    OrganizationRoleType,
    OrganizationType,
)
from warehouse.packaging import Project
from warehouse.utils.paginate import paginate_url_factory


class TestManageOrganizationApplication:
    def test_manage_organization_application(self, db_request):
        _organization_application = OrganizationApplicationFactory.create(
            status=OrganizationApplicationStatus.Submitted
        )

        view = org_views.ManageOrganizationApplicationViews(
            _organization_application, db_request
        )
        assert view.manage_organization_application() == {
            "organization_application": _organization_application,
            "information_requests": [],
            "response_forms": {},
        }
        assert (
            _organization_application.status == OrganizationApplicationStatus.Submitted
        )

    def test_manage_organization_application_response_with_info_requests(
        self, db_request, organization_service, monkeypatch
    ):
        _organization_application = OrganizationApplicationFactory.create(
            status=OrganizationApplicationStatus.MoreInformationNeeded
        )

        with freeze_time() as frozen_datetime:
            _request_for_more_info0 = OrganizationApplicationObservationFactory(
                related=_organization_application,
                payload={"message": "we need more information"},
                created=datetime.datetime.now(datetime.UTC),
            )
            frozen_datetime.tick(1.0)
            _request_for_more_info1 = OrganizationApplicationObservationFactory(
                related=_organization_application,
                payload={"message": "we still need more information"},
                created=datetime.datetime.now(datetime.UTC),
            )

        view = org_views.ManageOrganizationApplicationViews(
            _organization_application, db_request
        )
        _response = view.manage_organization_application()

        assert (
            _organization_application.status
            == OrganizationApplicationStatus.MoreInformationNeeded
        )
        assert _response["organization_application"] == _organization_application
        assert _response["information_requests"] == [
            _request_for_more_info1,
            _request_for_more_info0,
        ]
        assert len(_response["response_forms"]) == 2
        assert _request_for_more_info0.id in _response["response_forms"]
        assert _request_for_more_info1.id in _response["response_forms"]

    def test_manage_organization_application_response_with_info_requests_and_responses(
        self, db_request, organization_service, monkeypatch
    ):
        _organization_application = OrganizationApplicationFactory.create(
            status=OrganizationApplicationStatus.MoreInformationNeeded
        )

        with freeze_time() as frozen_datetime:
            _request_for_more_info0 = OrganizationApplicationObservationFactory(
                related=_organization_application,
                payload={"message": "we need more information"},
                additional={
                    "response": "you got it",
                    "response_time": "2025-03-14T15:40:36.485495+00:00",
                },
                created=datetime.datetime.now(datetime.UTC),
            )
            frozen_datetime.tick(1.0)
            _request_for_more_info1 = OrganizationApplicationObservationFactory(
                related=_organization_application,
                payload={"message": "we still need more information"},
                created=datetime.datetime.now(datetime.UTC),
            )

        view = org_views.ManageOrganizationApplicationViews(
            _organization_application, db_request
        )
        _response = view.manage_organization_application()

        assert (
            _organization_application.status
            == OrganizationApplicationStatus.MoreInformationNeeded
        )
        assert _response["organization_application"] == _organization_application
        assert _response["information_requests"] == [
            _request_for_more_info1,
            _request_for_more_info0,
        ]
        assert len(_response["response_forms"]) == 1
        assert _request_for_more_info1.id in _response["response_forms"]

    def test_manage_organization_application_response_with_all_responded(
        self, db_request, organization_service
    ):
        _organization_application = OrganizationApplicationFactory.create(
            status=OrganizationApplicationStatus.Submitted
        )

        with freeze_time() as frozen_datetime:
            _request_for_more_info0 = OrganizationApplicationObservationFactory(
                related=_organization_application,
                payload={"message": "we need more information"},
                additional={
                    "response": "you got it",
                    "response_time": "2025-03-14T15:40:36.485495+00:00",
                },
                created=datetime.datetime.now(datetime.UTC),
            )
            frozen_datetime.tick(1.0)
            _request_for_more_info1 = OrganizationApplicationObservationFactory(
                related=_organization_application,
                payload={"message": "we still need more information"},
                additional={
                    "response": "you got it",
                    "response_time": "2025-03-14T15:40:36.485495+00:00",
                },
                created=datetime.datetime.now(datetime.UTC),
            )

        view = org_views.ManageOrganizationApplicationViews(
            _organization_application, db_request
        )
        _response = view.manage_organization_application()

        assert (
            _organization_application.status == OrganizationApplicationStatus.Submitted
        )
        assert _response["organization_application"] == _organization_application
        assert _response["information_requests"] == [
            _request_for_more_info1,
            _request_for_more_info0,
        ]
        assert len(_response["response_forms"]) == 0

    def test_manage_organization_application_submit_empty(
        self, db_request, organization_service
    ):
        _organization_application = OrganizationApplicationFactory.create(
            status=OrganizationApplicationStatus.MoreInformationNeeded
        )

        with freeze_time() as frozen_datetime:
            _request_for_more_info0 = OrganizationApplicationObservationFactory(
                related=_organization_application,
                payload={"message": "we need more information"},
                additional={
                    "response": "you got it",
                    "response_time": "2025-03-14T15:40:36.485495+00:00",
                },
                created=datetime.datetime.now(datetime.UTC),
            )
            frozen_datetime.tick(1.0)
            _request_for_more_info1 = OrganizationApplicationObservationFactory(
                related=_organization_application,
                payload={"message": "we still need more information"},
                created=datetime.datetime.now(datetime.UTC),
            )

        db_request.POST = MultiDict({})

        view = org_views.ManageOrganizationApplicationViews(
            _organization_application, db_request
        )
        _response = view.manage_organization_application_submit()

        assert (
            _organization_application.status
            == OrganizationApplicationStatus.MoreInformationNeeded
        )
        assert _response["organization_application"] == _organization_application
        assert _response["information_requests"] == [
            _request_for_more_info1,
            _request_for_more_info0,
        ]
        assert len(_response["response_forms"]) == 1
        assert _request_for_more_info1.id in _response["response_forms"]

    def test_manage_organization_application_submit_response(
        self, db_request, organization_service
    ):
        _organization_application = OrganizationApplicationFactory.create(
            status=OrganizationApplicationStatus.MoreInformationNeeded
        )

        with freeze_time() as frozen_datetime:
            OrganizationApplicationObservationFactory(
                related=_organization_application,
                payload={"message": "we need more information"},
                additional={
                    "response": "you got it",
                    "response_time": "2025-03-14T15:40:36.485495+00:00",
                },
                created=datetime.datetime.now(datetime.UTC),
            )
            frozen_datetime.tick(1.0)
            _request_for_more_info1 = OrganizationApplicationObservationFactory(
                related=_organization_application,
                payload={"message": "we still need more information"},
                created=datetime.datetime.now(datetime.UTC),
            )

        db_request.POST = MultiDict(
            {
                "response_form-id": _request_for_more_info1.id.__str__(),
                "response": "This is my response",
            }
        )

        view = org_views.ManageOrganizationApplicationViews(
            _organization_application, db_request
        )
        _response = view.manage_organization_application_submit()

        assert (
            _organization_application.status == OrganizationApplicationStatus.Submitted
        )
        assert isinstance(_response, HTTPSeeOther)
        assert _request_for_more_info1.additional["response"] == "This is my response"

    def test_manage_organization_application_submit_response_correct_request(
        self, db_request, organization_service
    ):
        _organization_application = OrganizationApplicationFactory.create(
            status=OrganizationApplicationStatus.MoreInformationNeeded
        )

        with freeze_time() as frozen_datetime:
            OrganizationApplicationObservationFactory(
                related=_organization_application,
                payload={"message": "we need more information"},
                created=datetime.datetime.now(datetime.UTC),
            )
            frozen_datetime.tick(1.0)
            _request_for_more_info1 = OrganizationApplicationObservationFactory(
                related=_organization_application,
                payload={"message": "we still need more information"},
                created=datetime.datetime.now(datetime.UTC),
            )

        db_request.POST = MultiDict(
            {
                "response_form-id": _request_for_more_info1.id.__str__(),
                "response": "This is my response",
            }
        )

        view = org_views.ManageOrganizationApplicationViews(
            _organization_application, db_request
        )
        _response = view.manage_organization_application_submit()

        assert (
            _organization_application.status
            == OrganizationApplicationStatus.MoreInformationNeeded
        )
        assert isinstance(_response, HTTPSeeOther)
        assert _request_for_more_info1.additional["response"] == "This is my response"


class TestManageOrganizations:
    def test_default_response(self, monkeypatch):
        create_organization_application_obj = pretend.stub()
        create_organization_application_cls = pretend.call_recorder(
            lambda *a, **kw: create_organization_application_obj
        )
        monkeypatch.setattr(
            org_views,
            "CreateOrganizationApplicationForm",
            create_organization_application_cls,
        )

        organization = pretend.stub(name=pretend.stub(), is_approved=None)

        user_organizations = pretend.call_recorder(
            lambda *a, **kw: {
                "organizations_managed": [],
                "organizations_owned": [organization],
                "organizations_billing": [],
            }
        )
        monkeypatch.setattr(org_views, "user_organizations", user_organizations)

        organization_service = pretend.stub(
            get_organizations_by_user=lambda *a, **kw: [organization],
            get_organization_invites_by_user=lambda *a, **kw: [],
        )
        user_service = pretend.stub()
        request = pretend.stub(
            user=pretend.stub(
                id=pretend.stub(), username=pretend.stub(), organization_applications=[]
            ),
            find_service=lambda interface, **kw: {
                IOrganizationService: organization_service,
                IUserService: user_service,
            }[interface],
            registry=pretend.stub(
                settings={
                    "warehouse.organizations.max_undecided_organization_applications": 3
                }
            ),
        )

        view = org_views.ManageOrganizationsViews(request)

        assert view.default_response == {
            "organization_applications": [],
            "organization_invites": [],
            "organizations": [organization],
            "organizations_managed": [],
            "organizations_owned": [organization.name],
            "organizations_billing": [],
            "create_organization_application_form": create_organization_application_obj,
        }

    def test_manage_organizations(self, monkeypatch):
        request = pretend.stub(
            find_service=lambda *a, **kw: pretend.stub(),
            organization_access=True,
            user=pretend.stub(),
        )

        default_response = MultiDict({"default": "response"})
        monkeypatch.setattr(
            org_views.ManageOrganizationsViews,
            "default_response",
            default_response,
        )
        view = org_views.ManageOrganizationsViews(request)
        result = view.manage_organizations()

        assert result == default_response

    def test_manage_organizations_disable_organizations(self):
        request = pretend.stub(
            find_service=lambda *a, **kw: pretend.stub(),
            organization_access=False,
        )

        view = org_views.ManageOrganizationsViews(request)
        with pytest.raises(HTTPNotFound):
            view.manage_organizations()

    @pytest.mark.usefixtures("_enable_organizations")
    def test_create_organization_application(self, monkeypatch):
        admins = []
        user_service = pretend.stub(
            get_admins=pretend.call_recorder(lambda *a, **kw: admins),
        )

        organization = pretend.stub(
            id=pretend.stub(),
            name="psf",
            display_name="Python Software Foundation",
            orgtype="Community",
            link_url="https://www.python.org/psf/",
            description=(
                "To promote, protect, and advance the Python programming "
                "language, and to support and facilitate the growth of a "
                "diverse and international community of Python programmers"
            ),
            is_active=False,
            is_approved=None,
            record_event=pretend.call_recorder(lambda *a, **kw: None),
        )
        catalog_entry = pretend.stub()
        role = pretend.stub()
        organization_service = pretend.stub(
            add_organization_application=pretend.call_recorder(
                lambda *a, **kw: organization
            ),
            add_catalog_entry=pretend.call_recorder(lambda *a, **kw: catalog_entry),
            add_organization_role=pretend.call_recorder(lambda *a, **kw: role),
        )

        request = pretend.stub(
            POST={
                "name": organization.name,
                "display_name": organization.display_name,
                "orgtype": organization.orgtype,
                "link_url": organization.link_url,
                "description": organization.description,
            },
            domain=pretend.stub(),
            user=pretend.stub(
                id=pretend.stub(),
                username=pretend.stub(),
                has_primary_verified_email=True,
                record_event=pretend.call_recorder(lambda *a, **kw: None),
            ),
            session=pretend.stub(flash=pretend.call_recorder(lambda *a, **kw: None)),
            find_service=lambda interface, **kw: {
                IUserService: user_service,
                IOrganizationService: organization_service,
            }[interface],
            organization_access=True,
            remote_addr="0.0.0.0",
            ip_address=pretend.stub(
                ip_address="0.0.0.0",
                hashed_ip_address="deadbeef",
                geoip_info={"country_code": "US"},
            ),
            path="request-path",
            registry=pretend.stub(
                settings={
                    "warehouse.organizations.max_undecided_organization_applications": 3
                }
            ),
        )

        create_organization_application_obj = pretend.stub(
            data=request.POST,
            orgtype=pretend.stub(data=request.POST["orgtype"]),
            validate=lambda: True,
        )
        create_organization_application_cls = pretend.call_recorder(
            lambda *a, **kw: create_organization_application_obj
        )
        monkeypatch.setattr(
            org_views,
            "CreateOrganizationApplicationForm",
            create_organization_application_cls,
        )

        send_email = pretend.call_recorder(lambda *a, **kw: None)
        monkeypatch.setattr(
            org_views, "send_new_organization_requested_email", send_email
        )

        default_response = {"default": "response"}
        monkeypatch.setattr(
            org_views.ManageOrganizationsViews,
            "default_response",
            default_response,
        )

        view = org_views.ManageOrganizationsViews(request)
        result = view.create_organization_application()

        assert user_service.get_admins.calls == []
        assert organization_service.add_organization_application.calls == [
            pretend.call(
                name=organization.name,
                display_name=organization.display_name,
                orgtype=organization.orgtype,
                link_url=organization.link_url,
                description=organization.description,
                submitted_by=request.user,
            )
        ]
        assert organization_service.add_organization_role.calls == []
        assert organization.record_event.calls == []
        assert request.user.record_event.calls == []
        assert send_email.calls == [
            pretend.call(
                request,
                request.user,
                organization_name=organization.name,
            ),
        ]
        assert isinstance(result, HTTPSeeOther)

    @pytest.mark.usefixtures("_enable_organizations")
    def test_create_organization_application_with_subscription(self, monkeypatch):
        admins = []
        user_service = pretend.stub(
            get_admins=pretend.call_recorder(lambda *a, **kw: admins),
        )

        organization = pretend.stub(
            id=pretend.stub(),
            name="psf",
            normalized_name="psf",
            display_name="Python Software Foundation",
            orgtype="Company",
            link_url="https://www.python.org/psf/",
            description=(
                "To promote, protect, and advance the Python programming "
                "language, and to support and facilitate the growth of a "
                "diverse and international community of Python programmers"
            ),
            is_active=False,
            is_approved=None,
            record_event=pretend.call_recorder(lambda *a, **kw: None),
        )
        catalog_entry = pretend.stub()
        role = pretend.stub()
        organization_service = pretend.stub(
            add_organization_application=pretend.call_recorder(
                lambda *a, **kw: organization
            ),
            add_catalog_entry=pretend.call_recorder(lambda *a, **kw: catalog_entry),
            add_organization_role=pretend.call_recorder(lambda *a, **kw: role),
        )

        request = pretend.stub(
            POST={
                "name": organization.name,
                "display_name": organization.display_name,
                "orgtype": organization.orgtype,
                "link_url": organization.link_url,
                "description": organization.description,
            },
            domain=pretend.stub(),
            user=pretend.stub(
                id=pretend.stub(),
                username=pretend.stub(),
                has_primary_verified_email=True,
                record_event=pretend.call_recorder(lambda *a, **kw: None),
            ),
            session=pretend.stub(flash=pretend.call_recorder(lambda *a, **kw: None)),
            find_service=lambda interface, **kw: {
                IUserService: user_service,
                IOrganizationService: organization_service,
            }[interface],
            organization_access=True,
            remote_addr="0.0.0.0",
            ip_address=pretend.stub(
                ip_address="0.0.0.0",
                hashed_ip_address="deadbeef",
                geoip_info={"country_code": "US"},
            ),
            route_path=lambda *a, **kw: "manage-subscription-url",
            path="request-path",
            registry=pretend.stub(
                settings={
                    "warehouse.organizations.max_undecided_organization_applications": 3
                }
            ),
        )

        create_organization_application_obj = pretend.stub(
            data=request.POST,
            orgtype=pretend.stub(data=request.POST["orgtype"]),
            validate=lambda: True,
        )
        create_organization_application_cls = pretend.call_recorder(
            lambda *a, **kw: create_organization_application_obj
        )
        monkeypatch.setattr(
            org_views,
            "CreateOrganizationApplicationForm",
            create_organization_application_cls,
        )

        send_email = pretend.call_recorder(lambda *a, **kw: None)
        monkeypatch.setattr(
            org_views, "send_new_organization_requested_email", send_email
        )

        default_response = {"default": "response"}
        monkeypatch.setattr(
            org_views.ManageOrganizationsViews,
            "default_response",
            default_response,
        )

        view = org_views.ManageOrganizationsViews(request)
        result = view.create_organization_application()

        assert user_service.get_admins.calls == []
        assert organization_service.add_organization_application.calls == [
            pretend.call(
                name=organization.name,
                display_name=organization.display_name,
                orgtype=organization.orgtype,
                link_url=organization.link_url,
                description=organization.description,
                submitted_by=request.user,
            )
        ]
        assert organization_service.add_organization_role.calls == []
        assert organization.record_event.calls == []
        assert request.user.record_event.calls == []
        assert send_email.calls == [
            pretend.call(
                request,
                request.user,
                organization_name=organization.name,
            ),
        ]
        assert isinstance(result, HTTPSeeOther)
        assert result.headers["Location"] == "request-path"

    def test_create_organization_application_validation_fails(self, monkeypatch):
        admins = []
        user_service = pretend.stub(
            get_admins=pretend.call_recorder(lambda *a, **kw: admins),
        )

        organization = pretend.stub(
            record_event=pretend.call_recorder(lambda *a, **kw: None),
        )
        catalog_entry = pretend.stub()
        role = pretend.stub()
        organization_service = pretend.stub(
            add_organization_application=pretend.call_recorder(
                lambda *a, **kw: organization
            ),
            add_catalog_entry=pretend.call_recorder(lambda *a, **kw: catalog_entry),
            add_organization_role=pretend.call_recorder(lambda *a, **kw: role),
        )

        request = pretend.stub(
            POST={
                "name": None,
                "display_name": None,
                "orgtype": None,
                "link_url": None,
                "description": None,
            },
            domain=pretend.stub(),
            user=pretend.stub(
                id=pretend.stub(),
                username=pretend.stub(),
                has_primary_verified_email=True,
            ),
            session=pretend.stub(flash=pretend.call_recorder(lambda *a, **kw: None)),
            find_service=lambda interface, **kw: {
                IUserService: user_service,
                IOrganizationService: organization_service,
            }[interface],
            organization_access=True,
            remote_addr="0.0.0.0",
            ip_address=pretend.stub(
                ip_address="0.0.0.0",
                hashed_ip_address="deadbeef",
                geoip_info={"country_code": "US"},
            ),
            registry=pretend.stub(
                settings={
                    "warehouse.organizations.max_undecided_organization_applications": 3
                }
            ),
        )

        create_organization_application_obj = pretend.stub(
            validate=lambda: False, data=request.POST
        )
        create_organization_application_cls = pretend.call_recorder(
            lambda *a, **kw: create_organization_application_obj
        )
        monkeypatch.setattr(
            org_views,
            "CreateOrganizationApplicationForm",
            create_organization_application_cls,
        )

        send_email = pretend.call_recorder(lambda *a, **kw: None)
        monkeypatch.setattr(
            org_views, "send_new_organization_requested_email", send_email
        )

        view = org_views.ManageOrganizationsViews(request)
        result = view.create_organization_application()

        assert user_service.get_admins.calls == []
        assert organization_service.add_organization_application.calls == []
        assert organization_service.add_catalog_entry.calls == []
        assert organization_service.add_organization_role.calls == []
        assert organization.record_event.calls == []
        assert send_email.calls == []
        assert result == {
            "create_organization_application_form": create_organization_application_obj
        }

    def test_create_organization_application_disable_organizations(self):
        request = pretend.stub(
            find_service=lambda *a, **kw: pretend.stub(),
            organization_access=False,
        )

        view = org_views.ManageOrganizationsViews(request)
        with pytest.raises(HTTPNotFound):
            view.create_organization_application()


class TestManageOrganizationSettings:
    @pytest.mark.usefixtures("_enable_organizations")
    def test_manage_organization(self, db_request, organization_service, monkeypatch):
        db_request.user = pretend.stub()
        organization = OrganizationFactory.create()
        OrganizationProjectFactory.create(
            organization=organization, project=ProjectFactory.create()
        )

        save_organization_obj = pretend.stub()
        save_organization_cls = pretend.call_recorder(
            lambda *a, **kw: save_organization_obj
        )
        monkeypatch.setattr(org_views, "SaveOrganizationForm", save_organization_cls)

        save_organization_name_obj = pretend.stub()
        save_organization_name_cls = pretend.call_recorder(
            lambda *a, **kw: save_organization_name_obj
        )
        monkeypatch.setattr(
            org_views, "SaveOrganizationNameForm", save_organization_name_cls
        )

        view = org_views.ManageOrganizationSettingsViews(organization, db_request)
        result = view.manage_organization()

        assert view.request == db_request
        assert view.organization_service == organization_service
        assert result == {
            "organization": organization,
            "save_organization_form": save_organization_obj,
            "save_organization_name_form": save_organization_name_obj,
            "active_projects": view.active_projects,
        }
        assert save_organization_cls.calls == [
            pretend.call(
                MultiDict(
                    {
                        "name": organization.name,
                        "display_name": organization.display_name,
                        "link_url": organization.link_url,
                        "description": organization.description,
                        "orgtype": organization.orgtype,
                    }
                )
            ),
        ]

    @pytest.mark.usefixtures("_enable_organizations")
    @pytest.mark.parametrize(
        ("orgtype", "has_customer"),
        [(orgtype, True) for orgtype in list(OrganizationType)]
        + [(orgtype, False) for orgtype in list(OrganizationType)],
    )
    def test_save_organization(
        self,
        db_request,
        pyramid_user,
        orgtype,
        has_customer,
        billing_service,
        organization_service,
        monkeypatch,
    ):
        organization = OrganizationFactory.create(orgtype=orgtype)
        customer = StripeCustomerFactory.create()
        if has_customer:
            OrganizationStripeCustomerFactory.create(
                organization=organization, customer=customer
            )
        db_request.POST = {
            "display_name": organization.display_name,
            "link_url": organization.link_url,
            "description": organization.description,
            "orgtype": organization.orgtype,
        }

        db_request.registry.settings["site.name"] = "PiePeaEye"

        monkeypatch.setattr(
            organization_service,
            "update_organization",
            pretend.call_recorder(lambda *a, **kw: None),
        )
        monkeypatch.setattr(
            billing_service,
            "update_customer",
            pretend.call_recorder(lambda stripe_customer_id, name, description: None),
        )

        save_organization_obj = pretend.stub(
            validate=lambda: True, data=db_request.POST
        )
        save_organization_cls = pretend.call_recorder(
            lambda *a, **kw: save_organization_obj
        )
        monkeypatch.setattr(org_views, "SaveOrganizationForm", save_organization_cls)

        send_email = pretend.call_recorder(lambda *a, **kw: None)
        monkeypatch.setattr(org_views, "send_organization_updated_email", send_email)
        monkeypatch.setattr(
            org_views, "organization_owners", lambda *a, **kw: [pyramid_user]
        )

        view = org_views.ManageOrganizationSettingsViews(organization, db_request)
        result = view.save_organization()

        assert isinstance(result, HTTPSeeOther)
        assert organization_service.update_organization.calls == [
            pretend.call(organization.id, **db_request.POST)
        ]
        assert billing_service.update_customer.calls == (
            [
                pretend.call(
                    customer.customer_id,
                    (
                        f"PiePeaEye Organization - {organization.display_name} "
                        f"({organization.name})"
                    ),
                    organization.description,
                )
            ]
            if has_customer
            else []
        )
        assert send_email.calls == [
            pretend.call(
                db_request,
                {pyramid_user},
                organization_name=organization.name,
                organization_display_name=organization.display_name,
                organization_link_url=organization.link_url,
                organization_description=organization.description,
                organization_orgtype=organization.orgtype,
                previous_organization_display_name=organization.display_name,
                previous_organization_link_url=organization.link_url,
                previous_organization_description=organization.description,
                previous_organization_orgtype=organization.orgtype,
            ),
        ]

    @pytest.mark.usefixtures("_enable_organizations")
    def test_save_organization_validation_fails(
        self, db_request, organization_service, monkeypatch
    ):
        organization = OrganizationFactory.create()
        db_request.POST = {
            "display_name": organization.display_name,
            "link_url": organization.link_url,
            "description": organization.description,
            "orgtype": organization.orgtype,
        }
        db_request.user = pretend.stub()

        monkeypatch.setattr(
            organization_service,
            "update_organization",
            pretend.call_recorder(lambda *a, **kw: None),
        )

        save_organization_obj = pretend.stub(
            validate=lambda: False, data=db_request.POST
        )
        save_organization_cls = pretend.call_recorder(
            lambda *a, **kw: save_organization_obj
        )
        monkeypatch.setattr(org_views, "SaveOrganizationForm", save_organization_cls)

        save_organization_name_obj = pretend.stub()
        save_organization_name_cls = pretend.call_recorder(
            lambda *a, **kw: save_organization_name_obj
        )
        monkeypatch.setattr(
            org_views, "SaveOrganizationNameForm", save_organization_name_cls
        )

        view = org_views.ManageOrganizationSettingsViews(organization, db_request)
        result = view.save_organization()

        assert result == {
            **view.default_response,
            "save_organization_form": save_organization_obj,
        }
        assert organization_service.update_organization.calls == []

    @pytest.mark.usefixtures("_enable_organizations")
    def test_save_organization_name_wrong_confirm(
        self, db_request, organization_service, monkeypatch
    ):
        organization = OrganizationFactory.create(name="foobar")
        db_request.POST = {
            "confirm_current_organization_name": organization.name.upper(),
            "name": "FooBar",
        }
        db_request.route_path = pretend.call_recorder(lambda *a, **kw: "/the-redirect")
        db_request.session = pretend.stub(
            flash=pretend.call_recorder(lambda *a, **kw: None)
        )

        view = org_views.ManageOrganizationSettingsViews(organization, db_request)
        with pytest.raises(HTTPSeeOther):
            view.save_organization_name()

        assert db_request.session.flash.calls == [
            pretend.call(
                (
                    "Could not rename organization - "
                    "'FOOBAR' is not the same as 'foobar'"
                ),
                queue="error",
            )
        ]

    @pytest.mark.usefixtures("_enable_organizations")
    def test_disable_save_organization_name(
        self,
        db_request,
        pyramid_user,
        user_service,
        monkeypatch,
    ):
        organization = OrganizationFactory.create(name="foobar")
        db_request.POST = {
            "confirm_current_organization_name": organization.name,
            "name": "FooBar",
        }
        db_request.route_path = pretend.call_recorder(
            lambda *a, organization_name, **kw: (
                f"/manage/organization/{organization_name}/settings/"
            )
        )

        admin = None
        monkeypatch.setattr(
            user_service,
            "get_admin_user",
            pretend.call_recorder(lambda *a, **kw: admin),
        )

        save_organization_obj = pretend.stub()
        save_organization_cls = pretend.call_recorder(
            lambda *a, **kw: save_organization_obj
        )
        monkeypatch.setattr(org_views, "SaveOrganizationForm", save_organization_cls)

        save_organization_name_obj = pretend.stub(
            validate=lambda: True, name=pretend.stub(data=db_request.POST["name"])
        )
        save_organization_name_cls = pretend.call_recorder(
            lambda *a, **kw: save_organization_name_obj
        )
        monkeypatch.setattr(
            org_views, "SaveOrganizationNameForm", save_organization_name_cls
        )

        send_email = pretend.call_recorder(lambda *a, **kw: None)

        view = org_views.ManageOrganizationSettingsViews(organization, db_request)
        result = view.save_organization_name()

        assert isinstance(result, HTTPSeeOther)
        assert result.headers["Location"] == (
            f"/manage/organization/{organization.normalized_name}/settings/"
        )
        assert send_email.calls == []

    # When support for renaming orgs is re-introduced
    # @pytest.mark.usefixtures("_enable_organizations")
    # def test_save_organization_name(
    #    self,
    #    db_request,
    #    pyramid_user,
    #    organization_service,
    #    user_service,
    #    monkeypatch,
    # ):
    #    organization = OrganizationFactory.create(name="foobar")
    #    db_request.POST = {
    #        "confirm_current_organization_name": organization.name,
    #        "name": "FooBar",
    #    }
    #    db_request.route_path = pretend.call_recorder(
    #        lambda *a, organization_name, **kw: (
    #            f"/manage/organization/{organization_name}/settings/"
    #        )
    #    )

    #    def rename_organization(organization_id, organization_name):
    #        organization.name = organization_name

    #    monkeypatch.setattr(
    #        organization_service,
    #        "rename_organization",
    #        pretend.call_recorder(rename_organization),
    #    )

    #    admin = None
    #    monkeypatch.setattr(
    #        user_service,
    #        "get_admin_user",
    #        pretend.call_recorder(lambda *a, **kw: admin),
    #    )

    #    save_organization_obj = pretend.stub()
    #    save_organization_cls = pretend.call_recorder(
    #        lambda *a, **kw: save_organization_obj
    #    )
    #    monkeypatch.setattr(org_views, "SaveOrganizationForm", save_organization_cls)

    #    save_organization_name_obj = pretend.stub(
    #        validate=lambda: True, name=pretend.stub(data=db_request.POST["name"])
    #    )
    #    save_organization_name_cls = pretend.call_recorder(
    #        lambda *a, **kw: save_organization_name_obj
    #    )
    #    monkeypatch.setattr(
    #        org_views, "SaveOrganizationNameForm", save_organization_name_cls
    #    )

    #    send_email = pretend.call_recorder(lambda *a, **kw: None)
    #    monkeypatch.setattr(org_views, "send_organization_renamed_email", send_email)
    #    monkeypatch.setattr(
    #        org_views, "organization_owners", lambda *a, **kw: [pyramid_user]
    #    )

    #    view = org_views.ManageOrganizationSettingsViews(organization, db_request)
    #    result = view.save_organization_name()

    #    assert isinstance(result, HTTPSeeOther)
    #    assert result.headers["Location"] == (
    #        f"/manage/organization/{organization.normalized_name}/settings/#modal-close"
    #    )
    #    assert organization_service.rename_organization.calls == [
    #         pretend.call(organization.id, "FooBar")
    #    ]
    #    assert send_email.calls == [
    #        pretend.call(
    #            db_request,
    #            {pyramid_user},
    #            organization_name="FooBar",
    #            previous_organization_name="foobar",
    #        ),
    #    ]

    # @pytest.mark.usefixtures("_enable_organizations")
    # def test_save_organization_name_validation_fails(
    #    self, db_request, organization_service, monkeypatch
    # ):
    #    organization = OrganizationFactory.create(name="foobar")
    #    db_request.POST = {
    #        "confirm_current_organization_name": organization.name,
    #        "name": "FooBar",
    #    }
    #    db_request.user = pretend.stub()

    #    def rename_organization(organization_id, organization_name):
    #        organization.name = organization_name

    #    monkeypatch.setattr(
    #        organization_service,
    #        "rename_organization",
    #        pretend.call_recorder(rename_organization),
    #    )

    #    save_organization_obj = pretend.stub()
    #    save_organization_cls = pretend.call_recorder(
    #        lambda *a, **kw: save_organization_obj
    #    )
    #    monkeypatch.setattr(org_views, "SaveOrganizationForm", save_organization_cls)

    #    save_organization_name_obj = pretend.stub(
    #        validate=lambda: False, errors=pretend.stub(values=lambda: ["Invalid"])
    #    )
    #    save_organization_name_cls = pretend.call_recorder(
    #        lambda *a, **kw: save_organization_name_obj
    #    )
    #    monkeypatch.setattr(
    #        org_views, "SaveOrganizationNameForm", save_organization_name_cls
    #    )

    #    view = org_views.ManageOrganizationSettingsViews(organization, db_request)
    #    result = view.save_organization_name()

    #    assert result == {
    #        **view.default_response,
    #        "save_organization_name_form": save_organization_name_obj,
    #    }
    #    assert organization_service.rename_organization.calls == []

    @pytest.mark.usefixtures("_enable_organizations")
    def test_delete_organization(
        self,
        db_request,
        pyramid_user,
        organization_service,
        user_service,
        monkeypatch,
    ):
        organization = OrganizationFactory.create()
        db_request.POST = {"confirm_organization_name": organization.name}
        db_request.route_path = pretend.call_recorder(
            lambda *a, **kw: "/manage/organizations/"
        )

        monkeypatch.setattr(
            organization_service,
            "delete_organization",
            pretend.call_recorder(lambda *a, **kw: None),
        )

        admin = None
        monkeypatch.setattr(
            user_service,
            "get_admin_user",
            pretend.call_recorder(lambda *a, **kw: admin),
        )

        send_email = pretend.call_recorder(lambda *a, **kw: None)
        monkeypatch.setattr(org_views, "send_organization_deleted_email", send_email)
        monkeypatch.setattr(
            org_views, "organization_owners", lambda *a, **kw: [pyramid_user]
        )

        view = org_views.ManageOrganizationSettingsViews(organization, db_request)
        result = view.delete_organization()

        assert isinstance(result, HTTPSeeOther)
        assert result.headers["Location"] == "/manage/organizations/"
        assert organization_service.delete_organization.calls == [
            pretend.call(organization.id)
        ]
        assert send_email.calls == [
            pretend.call(
                db_request,
                {pyramid_user},
                organization_name=organization.name,
            ),
        ]
        assert db_request.route_path.calls == [pretend.call("manage.organizations")]

    @pytest.mark.usefixtures("_enable_organizations")
    def test_delete_organization_with_active_projects(
        self,
        db_request,
        pyramid_user,
        organization_service,
        monkeypatch,
    ):
        organization = OrganizationFactory.create()
        OrganizationProjectFactory.create(
            organization=organization, project=ProjectFactory.create()
        )

        db_request.POST = {"confirm_organization_name": organization.name}
        db_request.route_path = pretend.call_recorder(
            lambda *a, **kw: "/manage/organizations/"
        )

        save_organization_obj = pretend.stub()
        save_organization_cls = pretend.call_recorder(
            lambda *a, **kw: save_organization_obj
        )
        monkeypatch.setattr(org_views, "SaveOrganizationForm", save_organization_cls)

        save_organization_name_obj = pretend.stub()
        save_organization_name_cls = pretend.call_recorder(
            lambda *a, **kw: save_organization_name_obj
        )
        monkeypatch.setattr(
            org_views, "SaveOrganizationNameForm", save_organization_name_cls
        )

        monkeypatch.setattr(
            organization_service,
            "delete_organization",
            pretend.call_recorder(lambda *a, **kw: None),
        )

        view = org_views.ManageOrganizationSettingsViews(organization, db_request)
        result = view.delete_organization()

        assert result == view.default_response
        assert organization_service.delete_organization.calls == []
        assert db_request.route_path.calls == []

    @pytest.mark.usefixtures("_enable_organizations")
    def test_delete_organization_with_subscriptions(
        self,
        db_request,
        pyramid_user,
        organization_service,
        user_service,
        monkeypatch,
    ):
        organization = OrganizationFactory.create()
        stripe_customer = StripeCustomerFactory.create()
        OrganizationStripeCustomerFactory.create(
            organization=organization, customer=stripe_customer
        )
        subscription = StripeSubscriptionFactory.create(customer=stripe_customer)
        OrganizationStripeSubscriptionFactory.create(
            organization=organization, subscription=subscription
        )

        db_request.POST = {"confirm_organization_name": organization.name}
        db_request.route_path = pretend.call_recorder(
            lambda *a, **kw: "/manage/organizations/"
        )

        monkeypatch.setattr(
            organization_service,
            "delete_organization",
            pretend.call_recorder(lambda *a, **kw: None),
        )

        admin = None
        monkeypatch.setattr(
            user_service,
            "get_admin_user",
            pretend.call_recorder(lambda *a, **kw: admin),
        )

        send_email = pretend.call_recorder(lambda *a, **kw: None)
        monkeypatch.setattr(org_views, "send_organization_deleted_email", send_email)
        monkeypatch.setattr(
            org_views, "organization_owners", lambda *a, **kw: [pyramid_user]
        )

        view = org_views.ManageOrganizationSettingsViews(organization, db_request)
        result = view.delete_organization()

        assert isinstance(result, HTTPSeeOther)
        assert result.headers["Location"] == "/manage/organizations/"
        assert organization_service.delete_organization.calls == [
            pretend.call(organization.id)
        ]
        assert send_email.calls == [
            pretend.call(
                db_request,
                {pyramid_user},
                organization_name=organization.name,
            ),
        ]
        assert db_request.route_path.calls == [pretend.call("manage.organizations")]


class TestManageOrganizationBillingViews:
    @pytest.fixture
    def organization(self):
        organization = OrganizationFactory.create()
        OrganizationStripeCustomerFactory.create(organization=organization)
        return organization

    @pytest.fixture
    def organization_no_customer(self):
        return OrganizationFactory.create()

    @pytest.fixture
    def subscription(self, organization):
        return StripeSubscriptionFactory.create(
            stripe_customer_id=organization.customer.customer_id
        )

    @pytest.fixture
    def organization_subscription(self, organization, subscription):
        return OrganizationStripeSubscriptionFactory.create(
            organization=organization, subscription=subscription
        )

    @pytest.fixture
    def subscription_price(self):
        return StripeSubscriptionPriceFactory.create()

    def test_customer_id(
        self,
        db_request,
        subscription_service,
        organization,
    ):
        billing_service = pretend.stub(
            create_customer=lambda *a, **kw: {"id": organization.customer.customer_id},
        )

        view = org_views.ManageOrganizationBillingViews(organization, db_request)
        view.billing_service = billing_service
        customer_id = view.customer_id

        assert customer_id == organization.customer.customer_id

    def test_customer_id_local_mock(
        self,
        db_request,
        billing_service,
        subscription_service,
        organization_no_customer,
    ):
        db_request.registry.settings["site.name"] = "PyPI"

        view = org_views.ManageOrganizationBillingViews(
            organization_no_customer, db_request
        )
        customer_id = view.customer_id

        assert customer_id.startswith("mockcus_")

    def test_disable_organizations(
        self,
        db_request,
        billing_service,
        subscription_service,
        organization,
    ):
        db_request.organization_access = False
        view = org_views.ManageOrganizationBillingViews(organization, db_request)

        with pytest.raises(HTTPNotFound):
            view.create_or_manage_subscription()

    @pytest.mark.usefixtures("_enable_organizations")
    def test_activate_subscription(
        self,
        db_request,
        organization,
        monkeypatch,
    ):
        organization_activate_billing_form_obj = pretend.stub()
        organization_activate_billing_form_cls = pretend.call_recorder(
            lambda *a, **kw: organization_activate_billing_form_obj
        )
        monkeypatch.setattr(
            org_views,
            "OrganizationActivateBillingForm",
            organization_activate_billing_form_cls,
        )
        db_request.POST = MultiDict()

        view = org_views.ManageOrganizationBillingViews(organization, db_request)

        result = view.activate_subscription()

        assert result == {
            "organization": organization,
            "form": organization_activate_billing_form_obj,
        }

    @pytest.mark.usefixtures("_enable_organizations")
    def test_post_activate_subscription_valid(
        self,
        db_request,
        organization,
        monkeypatch,
    ):
        db_request.method = "POST"
        db_request.POST = MultiDict({"terms_of_service_agreement": "1"})

        db_request.route_path = pretend.call_recorder(
            lambda *a, **kw: "mock-billing-url"
        )

        view = org_views.ManageOrganizationBillingViews(organization, db_request)

        result = view.activate_subscription()

        assert isinstance(result, HTTPSeeOther)
        assert result.headers["Location"] == "mock-billing-url"

    @pytest.mark.usefixtures("_enable_organizations")
    def test_post_activate_subscription_invalid(
        self,
        db_request,
        organization,
        monkeypatch,
    ):
        db_request.method = "POST"
        db_request.POST = MultiDict()

        view = org_views.ManageOrganizationBillingViews(organization, db_request)

        result = view.activate_subscription()

        assert result["organization"] == organization
        assert result["form"].terms_of_service_agreement.errors == [
            "Terms of Service must be accepted."
        ]

    @pytest.mark.usefixtures("_enable_organizations")
    def test_create_subscription(
        self,
        db_request,
        subscription_service,
        organization,
        subscription_price,
        monkeypatch,
    ):
        db_request.route_path = pretend.call_recorder(
            lambda *a, **kw: "mock-session-url"
        )

        # Stub for billing service is not instance of MockStripeBillingService.
        create_checkout_session = pretend.call_recorder(
            lambda *a, **kw: {"url": "session-url"}
        )

        billing_service = pretend.stub(
            create_checkout_session=create_checkout_session,
            create_customer=lambda *a, **kw: {"id": organization.customer.customer_id},
            sync_price=lambda *a, **kw: None,
            sync_product=lambda *a, **kw: None,
        )

        view = org_views.ManageOrganizationBillingViews(organization, db_request)
        view.billing_service = billing_service
        result = view.create_or_manage_subscription()

        assert create_checkout_session.calls == [
            pretend.call(
                customer_id=organization.customer.customer_id,
                price_ids=[subscription_price.price_id],
                success_url=view.return_url,
                cancel_url=view.return_url,
            ),
        ]
        assert isinstance(result, HTTPSeeOther)
        assert result.headers["Location"] == "session-url"

    @pytest.mark.usefixtures("_enable_organizations")
    def test_create_subscription_local_mock(
        self,
        db_request,
        billing_service,
        subscription_service,
        organization,
        subscription_price,
        monkeypatch,
    ):
        db_request.route_path = pretend.call_recorder(
            lambda *a, **kw: "mock-session-url"
        )

        # Fixture for billing service is instance of MockStripeBillingService.
        create_checkout_session = pretend.call_recorder(
            lambda *a, **kw: {"url": "session-url"}
        )
        monkeypatch.setattr(
            billing_service, "create_checkout_session", create_checkout_session
        )

        view = org_views.ManageOrganizationBillingViews(organization, db_request)
        result = view.create_or_manage_subscription()

        assert create_checkout_session.calls == [
            pretend.call(
                customer_id=view.customer_id,
                price_ids=[subscription_price.price_id],
                success_url=view.return_url,
                cancel_url=view.return_url,
            ),
        ]
        assert isinstance(result, HTTPSeeOther)
        assert result.headers["Location"] == "mock-session-url"

    @pytest.mark.usefixtures("_enable_organizations")
    def test_manage_subscription(
        self,
        db_request,
        billing_service,
        subscription_service,
        organization,
        organization_subscription,
        monkeypatch,
    ):
        db_request.route_path = pretend.call_recorder(
            lambda *a, **kw: "mock-session-url"
        )

        # Stub for billing service is not instance of MockStripeBillingService.
        create_portal_session = pretend.call_recorder(
            lambda *a, **kw: {"url": "session-url"}
        )
        billing_service = pretend.stub(
            create_portal_session=create_portal_session,
            sync_price=lambda *a, **kw: None,
            sync_product=lambda *a, **kw: None,
        )

        view = org_views.ManageOrganizationBillingViews(organization, db_request)
        view.billing_service = billing_service
        result = view.create_or_manage_subscription()

        assert create_portal_session.calls == [
            pretend.call(
                customer_id=organization.customer.customer_id,
                return_url=view.return_url,
            ),
        ]
        assert isinstance(result, HTTPSeeOther)
        assert result.headers["Location"] == "session-url"

    @pytest.mark.usefixtures("_enable_organizations")
    def test_manage_subscription_local_mock(
        self,
        db_request,
        billing_service,
        subscription_service,
        organization,
        organization_subscription,
        monkeypatch,
    ):
        db_request.route_path = pretend.call_recorder(
            lambda *a, **kw: "mock-session-url"
        )

        # Fixture for billing service is instance of MockStripeBillingService.
        create_portal_session = pretend.call_recorder(
            lambda *a, **kw: {"url": "session-url"}
        )
        monkeypatch.setattr(
            billing_service, "create_portal_session", create_portal_session
        )

        view = org_views.ManageOrganizationBillingViews(organization, db_request)
        result = view.create_or_manage_subscription()

        assert create_portal_session.calls == [
            pretend.call(
                customer_id=organization.customer.customer_id,
                return_url=view.return_url,
            ),
        ]
        assert isinstance(result, HTTPSeeOther)
        assert result.headers["Location"] == "mock-session-url"


class TestManageOrganizationTeams:
    @pytest.mark.usefixtures("_enable_organizations")
    def test_manage_teams(
        self,
        db_request,
        pyramid_user,
        organization_service,
        monkeypatch,
    ):
        organization = OrganizationFactory.create()
        organization.teams = [TeamFactory.create()]

        db_request.POST = MultiDict()

        view = org_views.ManageOrganizationTeamsViews(organization, db_request)
        result = view.manage_teams()
        form = result["create_team_form"]

        assert view.request == db_request
        assert view.organization_service == organization_service
        assert result == {
            "organization": organization,
            "create_team_form": form,
        }

    @pytest.mark.usefixtures("_enable_organizations")
    def test_create_team(
        self,
        db_request,
        pyramid_user,
        organization_service,
        monkeypatch,
    ):
        organization = OrganizationFactory.create()
        organization.teams = [TeamFactory.create()]

        db_request.POST = MultiDict({"name": "Team Name"})

        OrganizationRoleFactory.create(
            organization=organization, user=db_request.user, role_name="Owner"
        )

        def add_team(name, *args, **kwargs):
            team = TeamFactory.create(name=name)
            organization.teams.append(team)
            return team

        monkeypatch.setattr(organization_service, "add_team", add_team)

        send_team_created_email = pretend.call_recorder(lambda *a, **kw: None)
        monkeypatch.setattr(
            org_views,
            "send_team_created_email",
            send_team_created_email,
        )

        view = org_views.ManageOrganizationTeamsViews(organization, db_request)
        result = view.create_team()

        assert isinstance(result, HTTPSeeOther)
        assert result.headers["Location"] == db_request.path
        assert len(organization.teams) == 2
        assert organization.teams[-1].name == "Team Name"
        assert send_team_created_email.calls == [
            pretend.call(
                db_request,
                {db_request.user},
                organization_name=organization.name,
                team_name="Team Name",
            )
        ]

    @pytest.mark.usefixtures("_enable_organizations")
    def test_create_team_invalid(
        self,
        db_request,
        pyramid_user,
        organization_service,
        monkeypatch,
    ):
        organization = OrganizationFactory.create()
        organization.teams = [TeamFactory.create(name="Used Name")]

        OrganizationRoleFactory.create(
            organization=organization, user=db_request.user, role_name="Owner"
        )

        db_request.POST = MultiDict({"name": "Used Name"})

        view = org_views.ManageOrganizationTeamsViews(organization, db_request)
        result = view.create_team()
        form = result["create_team_form"]

        assert view.request == db_request
        assert view.organization_service == organization_service
        assert result == {
            "organization": organization,
            "create_team_form": form,
        }
        assert form.name.errors == [
            "This team name has already been used. Choose a different team name."
        ]
        assert len(organization.teams) == 1


class TestManageOrganizationProjects:
    @pytest.mark.usefixtures("_enable_organizations")
    def test_manage_organization_projects(
        self,
        db_request,
        pyramid_user,
        organization_service,
        monkeypatch,
    ):
        organization = OrganizationFactory.create()
        OrganizationProjectFactory.create(
            organization=organization, project=ProjectFactory.create()
        )

        add_organization_project_obj = pretend.stub()
        add_organization_project_cls = pretend.call_recorder(
            lambda *a, **kw: add_organization_project_obj
        )
        monkeypatch.setattr(
            org_views, "AddOrganizationProjectForm", add_organization_project_cls
        )

        view = org_views.ManageOrganizationProjectsViews(organization, db_request)
        result = view.manage_organization_projects()

        assert view.request == db_request
        assert view.organization_service == organization_service
        assert result == {
            "organization": organization,
            "active_projects": view.active_projects,
            "projects_owned": set(),
            "projects_sole_owned": set(),
            "add_organization_project_form": add_organization_project_obj,
        }
        assert len(add_organization_project_cls.calls) == 1

    @pytest.mark.usefixtures("_enable_organizations")
    def test_add_organization_project_existing_project(
        self,
        db_request,
        pyramid_user,
        organization_service,
        monkeypatch,
    ):
        organization = OrganizationFactory.create()
        OrganizationProjectFactory.create(
            organization=organization, project=ProjectFactory.create()
        )

        project = ProjectFactory.create()

        OrganizationRoleFactory.create(
            organization=organization, user=db_request.user, role_name="Owner"
        )
        RoleFactory.create(project=project, user=db_request.user, role_name="Owner")

        add_organization_project_obj = pretend.stub(
            add_existing_project=pretend.stub(data=True),
            existing_project_name=pretend.stub(data=project.name),
            validate=lambda *a, **kw: True,
        )
        add_organization_project_cls = pretend.call_recorder(
            lambda *a, **kw: add_organization_project_obj
        )
        monkeypatch.setattr(
            org_views, "AddOrganizationProjectForm", add_organization_project_cls
        )

        def add_organization_project(*args, **kwargs):
            OrganizationProjectFactory.create(
                organization=organization, project=project
            )

        monkeypatch.setattr(
            organization_service, "add_organization_project", add_organization_project
        )

        send_organization_project_added_email = pretend.call_recorder(
            lambda req, user, **k: None
        )
        monkeypatch.setattr(
            org_views,
            "send_organization_project_added_email",
            send_organization_project_added_email,
        )

        view = org_views.ManageOrganizationProjectsViews(organization, db_request)
        result = view.add_organization_project()

        assert isinstance(result, HTTPSeeOther)
        assert result.headers["Location"] == db_request.path
        assert len(add_organization_project_cls.calls) == 1
        assert len(organization.projects) == 2
        assert send_organization_project_added_email.calls == [
            pretend.call(
                db_request,
                {db_request.user},
                organization_name=organization.name,
                project_name=project.name,
            )
        ]

    @pytest.mark.usefixtures("_enable_organizations")
    def test_add_organization_project_existing_project_no_individual_owner(
        self,
        db_request,
        pyramid_user,
        organization_service,
        monkeypatch,
    ):
        organization = OrganizationFactory.create()
        OrganizationProjectFactory.create(
            organization=organization, project=ProjectFactory.create()
        )

        project = ProjectFactory.create()

        OrganizationRoleFactory.create(
            organization=organization, user=db_request.user, role_name="Owner"
        )

        add_organization_project_obj = pretend.stub(
            add_existing_project=pretend.stub(data=True),
            existing_project_name=pretend.stub(data=project.name),
            validate=lambda *a, **kw: True,
        )
        add_organization_project_cls = pretend.call_recorder(
            lambda *a, **kw: add_organization_project_obj
        )
        monkeypatch.setattr(
            org_views, "AddOrganizationProjectForm", add_organization_project_cls
        )

        def add_organization_project(*args, **kwargs):
            OrganizationProjectFactory.create(
                organization=organization, project=project
            )

        monkeypatch.setattr(
            organization_service, "add_organization_project", add_organization_project
        )

        send_organization_project_added_email = pretend.call_recorder(
            lambda req, user, **k: None
        )
        monkeypatch.setattr(
            org_views,
            "send_organization_project_added_email",
            send_organization_project_added_email,
        )

        view = org_views.ManageOrganizationProjectsViews(organization, db_request)
        result = view.add_organization_project()

        assert isinstance(result, HTTPSeeOther)
        assert result.headers["Location"] == db_request.path
        assert len(add_organization_project_cls.calls) == 1
        assert len(organization.projects) == 2
        assert send_organization_project_added_email.calls == [
            pretend.call(
                db_request,
                {db_request.user},
                organization_name=organization.name,
                project_name=project.name,
            )
        ]

    @pytest.mark.usefixtures("_enable_organizations")
    def test_add_organization_project_existing_project_invalid(
        self,
        db_request,
        pyramid_user,
        monkeypatch,
    ):
        organization = OrganizationFactory.create()
        OrganizationProjectFactory.create(
            organization=organization, project=ProjectFactory.create()
        )

        project = ProjectFactory.create()

        OrganizationRoleFactory.create(
            organization=organization, user=db_request.user, role_name="Owner"
        )
        RoleFactory.create(project=project, user=db_request.user, role_name="Owner")

        add_organization_project_obj = pretend.stub(
            add_existing_project=pretend.stub(data=True),
            existing_project_name=pretend.stub(data=project.name),
            validate=lambda *a, **kw: False,
        )
        add_organization_project_cls = pretend.call_recorder(
            lambda *a, **kw: add_organization_project_obj
        )
        monkeypatch.setattr(
            org_views, "AddOrganizationProjectForm", add_organization_project_cls
        )

        view = org_views.ManageOrganizationProjectsViews(organization, db_request)
        result = view.add_organization_project()

        assert result == {
            "organization": organization,
            "active_projects": view.active_projects,
            "projects_owned": {project.name, organization.projects[0].name},
            "projects_sole_owned": {project.name, organization.projects[0].name},
            "add_organization_project_form": add_organization_project_obj,
        }
        assert len(add_organization_project_cls.calls) == 1
        assert len(organization.projects) == 1

    @pytest.mark.usefixtures("_enable_organizations")
    def test_add_organization_project_new_project(
        self,
        db_request,
        pyramid_user,
        monkeypatch,
    ):
        db_request.help_url = lambda *a, **kw: ""

        organization = OrganizationFactory.create()

        OrganizationRoleFactory.create(
            organization=organization, user=db_request.user, role_name="Owner"
        )

        add_organization_project_obj = pretend.stub(
            add_existing_project=pretend.stub(data=False),
            new_project_name=pretend.stub(data="fakepackage"),
            validate=lambda *a, **kw: True,
        )
        add_organization_project_cls = pretend.call_recorder(
            lambda *a, **kw: add_organization_project_obj
        )
        monkeypatch.setattr(
            org_views, "AddOrganizationProjectForm", add_organization_project_cls
        )

        send_organization_project_added_email = pretend.call_recorder(
            lambda req, user, **k: None
        )
        monkeypatch.setattr(
            org_views,
            "send_organization_project_added_email",
            send_organization_project_added_email,
        )

        view = org_views.ManageOrganizationProjectsViews(organization, db_request)
        result = view.add_organization_project()

        # The project was created
        project = db_request.db.query(Project).filter_by(name="fakepackage").one()

        # Refresh the project in the DB session to ensure it is not stale
        db_request.db.refresh(project)

        # The project belongs to the organization.
        assert project.organization == organization

        assert isinstance(result, HTTPSeeOther)
        assert result.headers["Location"] == db_request.path
        assert send_organization_project_added_email.calls == [
            pretend.call(
                db_request,
                {db_request.user},
                organization_name=organization.name,
                project_name="fakepackage",
            )
        ]

    @pytest.mark.usefixtures("_enable_organizations")
    @pytest.mark.parametrize(
        ("invalid_name", "expected"),
        [
            ("-invalid-name-", "The name '-invalid-name-' is invalid."),
            (
                "requirements.txt",
                (
                    "The name 'requirements.txt' isn't allowed. See help-url "
                    "for more information."
                ),
            ),
        ],
    )
    def test_add_organization_project_new_project_exception(
        self,
        db_request,
        pyramid_user,
        organization_service,
        monkeypatch,
        invalid_name,
        expected,
    ):
        db_request.help_url = lambda *a, **kw: "help-url"

        organization = OrganizationFactory.create()
        OrganizationProjectFactory.create(
            organization=organization, project=ProjectFactory.create()
        )

        project = ProjectFactory.create()

        OrganizationRoleFactory.create(
            organization=organization, user=db_request.user, role_name="Owner"
        )
        RoleFactory.create(project=project, user=db_request.user, role_name="Owner")

        add_organization_project_obj = pretend.stub(
            add_existing_project=pretend.stub(data=False),
            new_project_name=pretend.stub(data=invalid_name, errors=[]),
            validate=lambda *a, **kw: True,
        )
        add_organization_project_cls = pretend.call_recorder(
            lambda *a, **kw: add_organization_project_obj
        )
        monkeypatch.setattr(
            org_views, "AddOrganizationProjectForm", add_organization_project_cls
        )

        view = org_views.ManageOrganizationProjectsViews(organization, db_request)
        result = view.add_organization_project()

        assert result == {
            "organization": organization,
            "active_projects": view.active_projects,
            "projects_owned": {project.name, organization.projects[0].name},
            "projects_sole_owned": {project.name, organization.projects[0].name},
            "add_organization_project_form": add_organization_project_obj,
        }
        assert add_organization_project_obj.new_project_name.errors == [expected]
        assert len(organization.projects) == 1

    @pytest.mark.usefixtures("_enable_organizations")
    def test_add_organization_project_new_project_name_conflict(
        self,
        db_request,
        pyramid_user,
        organization_service,
        monkeypatch,
    ):
        db_request.help_url = lambda *a, **kw: "help-url"

        organization = OrganizationFactory.create()
        OrganizationProjectFactory.create(
            organization=organization, project=ProjectFactory.create()
        )

        project = ProjectFactory.create()

        OrganizationRoleFactory.create(
            organization=organization, user=db_request.user, role_name="Owner"
        )
        RoleFactory.create(project=project, user=db_request.user, role_name="Owner")

        add_organization_project_obj = pretend.stub(
            add_existing_project=pretend.stub(data=False),
            new_project_name=pretend.stub(data=project.name, errors=[]),
            validate=lambda *a, **kw: True,
        )
        add_organization_project_cls = pretend.call_recorder(
            lambda *a, **kw: add_organization_project_obj
        )
        monkeypatch.setattr(
            org_views, "AddOrganizationProjectForm", add_organization_project_cls
        )

        view = org_views.ManageOrganizationProjectsViews(organization, db_request)
        result = view.add_organization_project()

        assert result == {
            "organization": organization,
            "active_projects": view.active_projects,
            "projects_owned": {project.name, organization.projects[0].name},
            "projects_sole_owned": {project.name, organization.projects[0].name},
            "add_organization_project_form": add_organization_project_obj,
        }
        assert add_organization_project_obj.new_project_name.errors == [
            (
                "The name {name!r} conflicts with an existing project. "
                "See {projecthelp} for more information."
            ).format(
                name=project.name,
                projecthelp="help-url",
            )
        ]
        assert len(organization.projects) == 1


class TestManageOrganizationRoles:
    @pytest.mark.usefixtures("_enable_organizations")
    def test_get_manage_organization_roles(self, db_request, pyramid_user):
        organization = OrganizationFactory.create(name="foobar")
        form_obj = pretend.stub()

        def form_class(*a, **kw):
            return form_obj

        result = org_views.manage_organization_roles(
            organization, db_request, _form_class=form_class
        )
        assert result == {
            "organization": organization,
            "roles": set(),
            "invitations": set(),
            "form": form_obj,
            "organizations_with_sole_owner": [],
        }

    @pytest.mark.usefixtures("_enable_organizations")
    @freeze_time(datetime.datetime.now(datetime.UTC))
    @pytest.mark.parametrize("orgtype", list(OrganizationType))
    def test_post_new_organization_role(
        self,
        db_request,
        orgtype,
        organization_service,
        user_service,
        token_service,
        monkeypatch,
    ):
        organization = OrganizationFactory.create(name="foobar", orgtype=orgtype)

        # Company organizations need billing to be in good standing
        if orgtype == OrganizationType.Company:
            OrganizationStripeCustomerFactory.create(organization=organization)
            OrganizationStripeSubscriptionFactory.create(organization=organization)

        new_user = UserFactory.create(username="new_user")
        EmailFactory.create(user=new_user, verified=True, primary=True)
        owner_1 = UserFactory.create(username="owner_1")
        owner_2 = UserFactory.create(username="owner_2")
        OrganizationRoleFactory.create(
            organization=organization,
            user=owner_1,
            role_name=OrganizationRoleType.Owner,
        )
        OrganizationRoleFactory.create(
            organization=organization,
            user=owner_2,
            role_name=OrganizationRoleType.Owner,
        )

        db_request.method = "POST"
        db_request.POST = MultiDict(
            {"username": new_user.username, "role_name": "Owner"}
        )
        db_request.user = owner_1
        db_request.session = pretend.stub(
            flash=pretend.call_recorder(lambda *a, **kw: None)
        )

        send_organization_member_invited_email = pretend.call_recorder(
            lambda r, u, **k: None
        )
        monkeypatch.setattr(
            org_views,
            "send_organization_member_invited_email",
            send_organization_member_invited_email,
        )
        send_organization_role_verification_email = pretend.call_recorder(
            lambda r, u, **k: None
        )
        monkeypatch.setattr(
            org_views,
            "send_organization_role_verification_email",
            send_organization_role_verification_email,
        )

        result = org_views.manage_organization_roles(organization, db_request)

        assert db_request.session.flash.calls == [
            pretend.call(f"Invitation sent to '{new_user.username}'", queue="success")
        ]

        # Only one role invitation is created
        (
            db_request.db.query(OrganizationInvitation)
            .filter(OrganizationInvitation.user == new_user)
            .filter(OrganizationInvitation.organization == organization)
            .one()
        )

        assert isinstance(result, HTTPSeeOther)
        assert send_organization_member_invited_email.calls == [
            pretend.call(
                db_request,
                {owner_1, owner_2},
                user=new_user,
                desired_role=db_request.POST["role_name"],
                initiator_username=db_request.user.username,
                organization_name=organization.name,
                email_token=token_service.dumps(
                    {
                        "action": "email-organization-role-verify",
                        "desired_role": db_request.POST["role_name"],
                        "user_id": new_user.id,
                        "organization_id": organization.id,
                        "submitter_id": db_request.user.id,
                    }
                ),
                token_age=token_service.max_age,
            )
        ]
        assert send_organization_role_verification_email.calls == [
            pretend.call(
                db_request,
                new_user,
                desired_role=db_request.POST["role_name"],
                initiator_username=db_request.user.username,
                organization_name=organization.name,
                email_token=token_service.dumps(
                    {
                        "action": "email-organization-role-verify",
                        "desired_role": db_request.POST["role_name"],
                        "user_id": new_user.id,
                        "organization_id": organization.id,
                        "submitter_id": db_request.user.id,
                    }
                ),
                token_age=token_service.max_age,
            )
        ]

    @pytest.mark.usefixtures("_enable_organizations")
    def test_post_duplicate_organization_role(
        self, db_request, organization_service, user_service
    ):
        organization = OrganizationFactory.create(name="foobar")
        user = UserFactory.create(username="testuser")
        role = OrganizationRoleFactory.create(
            organization=organization,
            user=user,
            role_name=OrganizationRoleType.Owner,
        )

        db_request.method = "POST"
        db_request.POST = pretend.stub()
        db_request.session = pretend.stub(
            flash=pretend.call_recorder(lambda *a, **kw: None)
        )
        form_obj = pretend.stub(
            validate=pretend.call_recorder(lambda: True),
            username=pretend.stub(data=user.username),
            role_name=pretend.stub(data=role.role_name),
        )
        form_class = pretend.call_recorder(lambda *a, **kw: form_obj)

        result = org_views.manage_organization_roles(
            organization, db_request, _form_class=form_class
        )

        assert form_obj.validate.calls == [pretend.call()]
        assert form_class.calls == [
            pretend.call(
                db_request.POST,
                orgtype=organization.orgtype,
                organization_service=organization_service,
                user_service=user_service,
            ),
        ]
        assert db_request.session.flash.calls == [
            pretend.call(
                "User 'testuser' already has Owner role for organization", queue="error"
            )
        ]

        # No additional roles are created
        assert role == db_request.db.query(OrganizationRole).one()

        assert isinstance(result, HTTPSeeOther)

    @pytest.mark.usefixtures("_enable_organizations")
    @pytest.mark.parametrize("with_email", [True, False])
    def test_post_unverified_email(
        self,
        db_request,
        organization_service,
        user_service,
        with_email,
    ):
        organization = OrganizationFactory.create(name="foobar")
        user = UserFactory.create(username="testuser")
        if with_email:
            EmailFactory.create(user=user, verified=False, primary=True)

        db_request.method = "POST"
        db_request.POST = pretend.stub()
        db_request.session = pretend.stub(
            flash=pretend.call_recorder(lambda *a, **kw: None)
        )
        form_obj = pretend.stub(
            validate=pretend.call_recorder(lambda: True),
            username=pretend.stub(data=user.username),
            role_name=pretend.stub(data=OrganizationRoleType.Owner),
        )
        form_class = pretend.call_recorder(lambda *a, **kw: form_obj)

        result = org_views.manage_organization_roles(
            organization, db_request, _form_class=form_class
        )

        assert form_obj.validate.calls == [pretend.call()]
        assert form_class.calls == [
            pretend.call(
                db_request.POST,
                orgtype=organization.orgtype,
                organization_service=organization_service,
                user_service=user_service,
            ),
        ]
        assert db_request.session.flash.calls == [
            pretend.call(
                "User 'testuser' does not have a verified primary email address "
                "and cannot be added as a Owner for organization",
                queue="error",
            )
        ]

        # No additional roles are created
        assert db_request.db.query(OrganizationRole).all() == []

        assert isinstance(result, HTTPSeeOther)

    @pytest.mark.usefixtures("_enable_organizations")
    def test_cannot_reinvite_organization_role(
        self, db_request, organization_service, user_service
    ):
        organization = OrganizationFactory.create(name="foobar")
        new_user = UserFactory.create(username="new_user")
        EmailFactory.create(user=new_user, verified=True, primary=True)
        owner_1 = UserFactory.create(username="owner_1")
        owner_2 = UserFactory.create(username="owner_2")
        OrganizationRoleFactory.create(
            organization=organization,
            user=owner_1,
            role_name=OrganizationRoleType.Owner,
        )
        OrganizationRoleFactory.create(
            organization=organization,
            user=owner_2,
            role_name=OrganizationRoleType.Owner,
        )
        token_service = db_request.find_service(ITokenService, name="email")
        OrganizationInvitationFactory.create(
            organization=organization,
            user=new_user,
            invite_status=OrganizationInvitationStatus.Pending,
            token=token_service.dumps({"action": "email-organization-role-verify"}),
        )

        db_request.method = "POST"
        db_request.POST = pretend.stub()
        db_request.remote_addr = "10.10.10.10"
        db_request.user = owner_1
        db_request.session = pretend.stub(
            flash=pretend.call_recorder(lambda *a, **kw: None)
        )
        form_obj = pretend.stub(
            validate=pretend.call_recorder(lambda: True),
            username=pretend.stub(data=new_user.username),
            role_name=pretend.stub(data=OrganizationRoleType.Owner),
        )
        form_class = pretend.call_recorder(lambda *a, **kw: form_obj)

        result = org_views.manage_organization_roles(
            organization, db_request, _form_class=form_class
        )

        assert form_obj.validate.calls == [pretend.call()]
        assert form_class.calls == [
            pretend.call(
                db_request.POST,
                orgtype=organization.orgtype,
                organization_service=organization_service,
                user_service=user_service,
            ),
        ]
        assert db_request.session.flash.calls == [
            pretend.call(
                "User 'new_user' already has an active invite. Please try again later.",
                queue="error",
            )
        ]
        assert isinstance(result, HTTPSeeOther)

    @pytest.mark.usefixtures("_enable_organizations")
    @freeze_time(datetime.datetime.now(datetime.UTC))
    def test_reinvite_organization_role_after_expiration(
        self,
        db_request,
        organization_service,
        user_service,
        monkeypatch,
    ):
        organization = OrganizationFactory.create(name="foobar")
        new_user = UserFactory.create(username="new_user")
        EmailFactory.create(user=new_user, verified=True, primary=True)
        owner_1 = UserFactory.create(username="owner_1")
        owner_2 = UserFactory.create(username="owner_2")
        OrganizationRoleFactory.create(
            organization=organization,
            user=owner_1,
            role_name=OrganizationRoleType.Owner,
        )
        OrganizationRoleFactory.create(
            user=owner_2,
            organization=organization,
            role_name=OrganizationRoleType.Owner,
        )
        token_service = db_request.find_service(ITokenService, name="email")
        OrganizationInvitationFactory.create(
            user=new_user,
            organization=organization,
            invite_status=OrganizationInvitationStatus.Expired,
            token=token_service.dumps({}),
        )

        db_request.method = "POST"
        db_request.POST = pretend.stub()
        db_request.remote_addr = "10.10.10.10"
        db_request.user = owner_1
        db_request.session = pretend.stub(
            flash=pretend.call_recorder(lambda *a, **kw: None)
        )
        form_obj = pretend.stub(
            validate=pretend.call_recorder(lambda: True),
            username=pretend.stub(data=new_user.username),
            role_name=pretend.stub(data=OrganizationRoleType.Owner),
        )
        form_class = pretend.call_recorder(lambda *a, **kw: form_obj)

        send_organization_member_invited_email = pretend.call_recorder(
            lambda r, u, **k: None
        )
        monkeypatch.setattr(
            org_views,
            "send_organization_member_invited_email",
            send_organization_member_invited_email,
        )
        send_organization_role_verification_email = pretend.call_recorder(
            lambda r, u, **k: None
        )
        monkeypatch.setattr(
            org_views,
            "send_organization_role_verification_email",
            send_organization_role_verification_email,
        )

        result = org_views.manage_organization_roles(
            organization, db_request, _form_class=form_class
        )

        assert form_obj.validate.calls == [pretend.call()]
        assert form_class.calls == [
            pretend.call(
                db_request.POST,
                orgtype=organization.orgtype,
                organization_service=organization_service,
                user_service=user_service,
            ),
        ]
        assert db_request.session.flash.calls == [
            pretend.call(f"Invitation sent to '{new_user.username}'", queue="success")
        ]

        # Only one role invitation is created
        (
            db_request.db.query(OrganizationInvitation)
            .filter(OrganizationInvitation.user == new_user)
            .filter(OrganizationInvitation.organization == organization)
            .one()
        )

        assert isinstance(result, HTTPSeeOther)
        assert send_organization_member_invited_email.calls == [
            pretend.call(
                db_request,
                {owner_1, owner_2},
                user=new_user,
                desired_role=form_obj.role_name.data.value,
                initiator_username=db_request.user.username,
                organization_name=organization.name,
                email_token=token_service.dumps(
                    {
                        "action": "email-organization-role-verify",
                        "desired_role": form_obj.role_name.data.value,
                        "user_id": new_user.id,
                        "organization_id": organization.id,
                        "submitter_id": db_request.user.id,
                    }
                ),
                token_age=token_service.max_age,
            )
        ]
        assert send_organization_role_verification_email.calls == [
            pretend.call(
                db_request,
                new_user,
                desired_role=form_obj.role_name.data.value,
                initiator_username=db_request.user.username,
                organization_name=organization.name,
                email_token=token_service.dumps(
                    {
                        "action": "email-organization-role-verify",
                        "desired_role": form_obj.role_name.data.value,
                        "user_id": new_user.id,
                        "organization_id": organization.id,
                        "submitter_id": db_request.user.id,
                    }
                ),
                token_age=token_service.max_age,
            )
        ]

    @pytest.mark.usefixtures("_enable_organizations")
    def test_manage_organization_roles_not_in_good_standing(
        self, db_request, monkeypatch
    ):
        organization = OrganizationFactory.create(orgtype="Company")  # No billing
        new_user = UserFactory.create(username="new-user")
        EmailFactory.create(user=new_user, verified=True, primary=True)
        owner_user = UserFactory.create()
        OrganizationRoleFactory(
            user=owner_user,
            organization=organization,
            role_name=OrganizationRoleType.Owner,
        )

        form_obj = pretend.stub(
            username=pretend.stub(data="new-user"),
            role_name=pretend.stub(data=OrganizationRoleType.Member),
            validate=pretend.call_recorder(lambda: True),
        )
        form_class = pretend.call_recorder(lambda *a, **kw: form_obj)

        db_request.method = "POST"
        db_request.user = owner_user
        db_request.session.flash = pretend.call_recorder(lambda *a, **kw: None)

        result = org_views.manage_organization_roles(
            organization, db_request, _form_class=form_class
        )

        assert db_request.session.flash.calls == [
            pretend.call(
                "Cannot invite new member. Organization is not in good standing.",
                queue="error",
            )
        ]
        assert isinstance(result, HTTPSeeOther)  # Redirect due to inactive org


class TestResendOrganizationInvitations:
    @pytest.mark.usefixtures("_enable_organizations")
    @freeze_time(datetime.datetime.now(datetime.UTC))
    def test_resend_invitation(self, db_request, token_service, monkeypatch):
        organization = OrganizationFactory.create(name="foobar")
        user = UserFactory.create(username="testuser")
        EmailFactory.create(user=user, verified=True, primary=True)
        OrganizationInvitationFactory.create(
            organization=organization,
            user=user,
            invite_status=OrganizationInvitationStatus.Expired,
        )
        owner_user = UserFactory.create()
        OrganizationRoleFactory(
            user=owner_user,
            organization=organization,
            role_name=OrganizationRoleType.Owner,
        )

        send_organization_member_invited_email = pretend.call_recorder(
            lambda r, u, **k: None
        )
        monkeypatch.setattr(
            org_views,
            "send_organization_member_invited_email",
            send_organization_member_invited_email,
        )
        send_organization_role_verification_email = pretend.call_recorder(
            lambda r, u, **k: None
        )
        monkeypatch.setattr(
            org_views,
            "send_organization_role_verification_email",
            send_organization_role_verification_email,
        )

        db_request.method = "POST"
        db_request.POST = MultiDict({"user_id": user.id})
        db_request.remote_addr = "10.10.10.10"
        db_request.user = owner_user
        db_request.route_path = pretend.call_recorder(
            lambda *a, **kw: "/manage/organizations"
        )
        db_request.session = pretend.stub(
            flash=pretend.call_recorder(lambda *a, **kw: None)
        )
        token_service.loads = pretend.raiser(TokenExpired)
        token_service.unsafe_load_payload = pretend.call_recorder(
            lambda data: {
                "action": "email-organization-role-verify",
                "desired_role": "Manager",
                "user_id": user.id,
                "organization_id": organization.id,
                "submitter_id": owner_user.id,
            }
        )

        result = org_views.resend_organization_invitation(organization, db_request)
        db_request.db.flush()

        assert (
            db_request.db.query(OrganizationInvitation)
            .filter(OrganizationInvitation.user == user)
            .filter(OrganizationInvitation.organization == organization)
            .filter(
                OrganizationInvitation.invite_status
                == OrganizationInvitationStatus.Pending
            )
            .one()
        )
        assert db_request.session.flash.calls == [
            pretend.call(f"Invitation sent to '{user.username}'", queue="success")
        ]

        assert isinstance(result, HTTPSeeOther)
        assert result.headers["Location"] == "/manage/organizations"

        assert send_organization_member_invited_email.calls == [
            pretend.call(
                db_request,
                {owner_user},
                user=user,
                desired_role="Manager",
                initiator_username=db_request.user.username,
                organization_name=organization.name,
                email_token=token_service.dumps(
                    {
                        "action": "email-organization-role-verify",
                        "desired_role": "Manager",
                        "user_id": user.id,
                        "organization_id": organization.id,
                        "submitter_id": db_request.user.id,
                    }
                ),
                token_age=token_service.max_age,
            )
        ]
        assert send_organization_role_verification_email.calls == [
            pretend.call(
                db_request,
                user,
                desired_role="Manager",
                initiator_username=db_request.user.username,
                organization_name=organization.name,
                email_token=token_service.dumps(
                    {
                        "action": "email-organization-role-verify",
                        "desired_role": "Manager",
                        "user_id": user.id,
                        "organization_id": organization.id,
                        "submitter_id": db_request.user.id,
                    }
                ),
                token_age=token_service.max_age,
            )
        ]

    @pytest.mark.usefixtures("_enable_organizations")
    def test_resend_invitation_fails_corrupt_token(self, db_request, token_service):
        organization = OrganizationFactory.create(name="foobar")
        user = UserFactory.create(username="testuser")
        OrganizationInvitationFactory.create(
            organization=organization,
            user=user,
            invite_status=OrganizationInvitationStatus.Expired,
        )
        owner_user = UserFactory.create()
        OrganizationRoleFactory(
            user=owner_user,
            organization=organization,
            role_name=OrganizationRoleType.Owner,
        )

        db_request.method = "POST"
        db_request.POST = MultiDict({"user_id": user.id})
        db_request.remote_addr = "10.10.10.10"
        db_request.user = owner_user
        db_request.route_path = pretend.call_recorder(
            lambda *a, **kw: "/manage/organizations"
        )
        db_request.session = pretend.stub(
            flash=pretend.call_recorder(lambda *a, **kw: None)
        )
        token_service.loads = pretend.raiser(TokenExpired)
        token_service.unsafe_load_payload = pretend.call_recorder(lambda data: None)

        result = org_views.resend_organization_invitation(organization, db_request)
        db_request.db.flush()

        assert (
            db_request.db.query(OrganizationInvitation)
            .filter(OrganizationInvitation.user == user)
            .filter(OrganizationInvitation.organization == organization)
            .filter(
                OrganizationInvitation.invite_status
                == OrganizationInvitationStatus.Pending
            )
            .one_or_none()
        ) is None
        assert db_request.session.flash.calls == [
            pretend.call("Organization invitation could not be re-sent.", queue="error")
        ]

        assert isinstance(result, HTTPSeeOther)
        assert result.headers["Location"] == "/manage/organizations"

    @pytest.mark.usefixtures("_enable_organizations")
    def test_resend_invitation_fails_missing_invitation(
        self, db_request, token_service
    ):
        organization = OrganizationFactory.create(name="foobar")
        user = UserFactory.create(username="testuser")
        owner_user = UserFactory.create()
        OrganizationRoleFactory(
            user=owner_user,
            organization=organization,
            role_name=OrganizationRoleType.Owner,
        )

        db_request.method = "POST"
        db_request.POST = MultiDict({"user_id": user.id})
        db_request.remote_addr = "10.10.10.10"
        db_request.user = owner_user
        db_request.route_path = pretend.call_recorder(
            lambda *a, **kw: "/manage/organizations"
        )
        db_request.session = pretend.stub(
            flash=pretend.call_recorder(lambda *a, **kw: None)
        )

        result = org_views.resend_organization_invitation(organization, db_request)
        db_request.db.flush()

        assert (
            db_request.db.query(OrganizationInvitation)
            .filter(OrganizationInvitation.user == user)
            .filter(OrganizationInvitation.organization == organization)
            .filter(
                OrganizationInvitation.invite_status
                == OrganizationInvitationStatus.Pending
            )
            .one_or_none()
        ) is None
        assert db_request.session.flash.calls == [
            pretend.call("Could not find organization invitation.", queue="error")
        ]

        assert isinstance(result, HTTPSeeOther)
        assert result.headers["Location"] == "/manage/organizations"


class TestRevokeOrganizationInvitation:
    @pytest.mark.usefixtures("_enable_organizations")
    def test_revoke_invitation(self, db_request, token_service, monkeypatch):
        organization = OrganizationFactory.create(name="foobar")
        user = UserFactory.create(username="testuser")
        OrganizationInvitationFactory.create(
            organization=organization,
            user=user,
        )
        owner_user = UserFactory.create()
        OrganizationRoleFactory(
            user=owner_user,
            organization=organization,
            role_name=OrganizationRoleType.Owner,
        )

        db_request.method = "POST"
        db_request.POST = MultiDict({"user_id": user.id, "token": "TOKEN"})
        db_request.remote_addr = "10.10.10.10"
        db_request.user = owner_user
        db_request.route_path = pretend.call_recorder(
            lambda *a, **kw: "/manage/organizations"
        )
        db_request.session = pretend.stub(
            flash=pretend.call_recorder(lambda *a, **kw: None)
        )
        token_service.loads = pretend.call_recorder(
            lambda data: {
                "action": "email-organization-role-verify",
                "desired_role": "Manager",
                "user_id": user.id,
                "organization_id": organization.id,
                "submitter_id": owner_user.id,
            }
        )

        organization_member_invite_canceled_email = pretend.call_recorder(
            lambda *args, **kwargs: None
        )
        monkeypatch.setattr(
            org_views,
            "send_organization_member_invite_canceled_email",
            organization_member_invite_canceled_email,
        )
        canceled_as_invited_organization_member_email = pretend.call_recorder(
            lambda *args, **kwargs: None
        )
        monkeypatch.setattr(
            org_views,
            "send_canceled_as_invited_organization_member_email",
            canceled_as_invited_organization_member_email,
        )

        result = org_views.revoke_organization_invitation(organization, db_request)
        db_request.db.flush()

        assert not (
            db_request.db.query(OrganizationInvitation)
            .filter(OrganizationInvitation.user == user)
            .filter(OrganizationInvitation.organization == organization)
            .one_or_none()
        )
        assert organization_member_invite_canceled_email.calls == [
            pretend.call(
                db_request,
                {owner_user},
                user=user,
                organization_name=organization.name,
            )
        ]
        assert canceled_as_invited_organization_member_email.calls == [
            pretend.call(
                db_request,
                user,
                organization_name=organization.name,
            )
        ]
        assert db_request.session.flash.calls == [
            pretend.call(f"Invitation revoked from '{user.username}'.", queue="success")
        ]
        assert isinstance(result, HTTPSeeOther)
        assert result.headers["Location"] == "/manage/organizations"

    @pytest.mark.usefixtures("_enable_organizations")
    def test_invitation_does_not_exist(self, db_request, token_service):
        organization = OrganizationFactory.create(name="foobar")
        user = UserFactory.create(username="testuser")
        owner_user = UserFactory.create()
        OrganizationRoleFactory(
            user=owner_user,
            organization=organization,
            role_name=OrganizationRoleType.Owner,
        )

        db_request.method = "POST"
        db_request.POST = MultiDict({"user_id": user.id, "token": "TOKEN"})
        db_request.remote_addr = "10.10.10.10"
        db_request.user = owner_user
        db_request.route_path = pretend.call_recorder(
            lambda *a, **kw: "/manage/organizations"
        )
        db_request.session = pretend.stub(
            flash=pretend.call_recorder(lambda *a, **kw: None)
        )
        token_service.loads = pretend.call_recorder(lambda data: None)

        result = org_views.revoke_organization_invitation(organization, db_request)
        db_request.db.flush()

        assert db_request.session.flash.calls == [
            pretend.call("Could not find organization invitation.", queue="error")
        ]
        assert isinstance(result, HTTPSeeOther)
        assert result.headers["Location"] == "/manage/organizations"

    @pytest.mark.usefixtures("_enable_organizations")
    def test_token_expired(self, db_request, token_service):
        organization = OrganizationFactory.create(name="foobar")
        user = UserFactory.create(username="testuser")
        OrganizationInvitationFactory.create(
            organization=organization,
            user=user,
        )
        owner_user = UserFactory.create()
        OrganizationRoleFactory(
            user=owner_user,
            organization=organization,
            role_name=OrganizationRoleType.Owner,
        )

        db_request.method = "POST"
        db_request.POST = MultiDict({"user_id": user.id, "token": "TOKEN"})
        db_request.remote_addr = "10.10.10.10"
        db_request.user = owner_user
        db_request.route_path = pretend.call_recorder(
            lambda *a, **kw: "/manage/organizations/roles"
        )
        db_request.session = pretend.stub(
            flash=pretend.call_recorder(lambda *a, **kw: None)
        )
        token_service.loads = pretend.call_recorder(pretend.raiser(TokenExpired))

        result = org_views.revoke_organization_invitation(organization, db_request)
        db_request.db.flush()

        assert not (
            db_request.db.query(OrganizationInvitation)
            .filter(OrganizationInvitation.user == user)
            .filter(OrganizationInvitation.organization == organization)
            .one_or_none()
        )
        assert db_request.session.flash.calls == [
            pretend.call("Expired invitation for 'testuser' deleted.", queue="success")
        ]
        assert isinstance(result, HTTPSeeOther)
        assert result.headers["Location"] == "/manage/organizations/roles"


class TestChangeOrganizationRole:
    @pytest.mark.usefixtures("_enable_organizations")
    @pytest.mark.parametrize("orgtype", list(OrganizationType))
    def test_change_role(self, db_request, orgtype, monkeypatch):
        organization = OrganizationFactory.create(name="foobar", orgtype=orgtype)
        user = UserFactory.create(username="testuser")
        role = OrganizationRoleFactory.create(
            organization=organization,
            user=user,
            role_name=OrganizationRoleType.Owner,
        )
        new_role_name = "Manager"

        user_2 = UserFactory.create()

        db_request.method = "POST"
        db_request.POST = MultiDict({"role_id": role.id, "role_name": new_role_name})
        db_request.user = user_2
        db_request.session = pretend.stub(
            flash=pretend.call_recorder(lambda *a, **kw: None)
        )
        db_request.route_path = pretend.call_recorder(lambda *a, **kw: "/the-redirect")

        send_organization_member_role_changed_email = pretend.call_recorder(
            lambda *a, **kw: None
        )
        monkeypatch.setattr(
            org_views,
            "send_organization_member_role_changed_email",
            send_organization_member_role_changed_email,
        )
        send_role_changed_as_organization_member_email = pretend.call_recorder(
            lambda *a, **kw: None
        )
        monkeypatch.setattr(
            org_views,
            "send_role_changed_as_organization_member_email",
            send_role_changed_as_organization_member_email,
        )

        result = org_views.change_organization_role(organization, db_request)

        assert role.role_name == new_role_name
        assert db_request.route_path.calls == [
            pretend.call(
                "manage.organization.roles", organization_name=organization.name
            )
        ]
        assert send_organization_member_role_changed_email.calls == [
            pretend.call(
                db_request,
                set(),
                user=user,
                submitter=user_2,
                organization_name="foobar",
                role=new_role_name,
            )
        ]
        assert send_role_changed_as_organization_member_email.calls == [
            pretend.call(
                db_request,
                user,
                submitter=user_2,
                organization_name="foobar",
                role=new_role_name,
            )
        ]
        assert db_request.session.flash.calls == [
            pretend.call("Changed role", queue="success")
        ]
        assert isinstance(result, HTTPSeeOther)
        assert result.headers["Location"] == "/the-redirect"

    @pytest.mark.usefixtures("_enable_organizations")
    def test_change_organization_role_invalid_role_name(self, db_request):
        organization = OrganizationFactory.create(name="foobar")

        db_request.method = "POST"
        db_request.POST = MultiDict(
            {"role_id": str(uuid.uuid4()), "role_name": "Invalid Role Name"}
        )
        db_request.route_path = pretend.call_recorder(lambda *a, **kw: "/the-redirect")

        result = org_views.change_organization_role(organization, db_request)

        assert db_request.route_path.calls == [
            pretend.call(
                "manage.organization.roles", organization_name=organization.name
            )
        ]
        assert isinstance(result, HTTPSeeOther)
        assert result.headers["Location"] == "/the-redirect"

    @pytest.mark.usefixtures("_enable_organizations")
    def test_change_missing_organization_role(self, db_request):
        organization = OrganizationFactory.create(name="foobar")
        missing_role_id = str(uuid.uuid4())

        db_request.method = "POST"
        db_request.POST = MultiDict({"role_id": missing_role_id, "role_name": "Owner"})
        db_request.user = pretend.stub()
        db_request.session = pretend.stub(
            flash=pretend.call_recorder(lambda *a, **kw: None)
        )
        db_request.route_path = pretend.call_recorder(lambda *a, **kw: "/the-redirect")

        result = org_views.change_organization_role(organization, db_request)

        assert db_request.session.flash.calls == [
            pretend.call("Could not find member", queue="error")
        ]
        assert isinstance(result, HTTPSeeOther)
        assert result.headers["Location"] == "/the-redirect"

    @pytest.mark.usefixtures("_enable_organizations")
    def test_change_own_owner_organization_role(self, db_request):
        organization = OrganizationFactory.create(name="foobar")
        user = UserFactory.create(username="testuser")
        role = OrganizationRoleFactory.create(
            user=user, organization=organization, role_name="Owner"
        )

        db_request.method = "POST"
        db_request.user = user
        db_request.POST = MultiDict({"role_id": role.id, "role_name": "Manager"})
        db_request.session = pretend.stub(
            flash=pretend.call_recorder(lambda *a, **kw: None)
        )
        db_request.route_path = pretend.call_recorder(lambda *a, **kw: "/the-redirect")

        result = org_views.change_organization_role(organization, db_request)

        assert db_request.session.flash.calls == [
            pretend.call("Cannot remove yourself as Owner", queue="error")
        ]
        assert isinstance(result, HTTPSeeOther)
        assert result.headers["Location"] == "/the-redirect"


class TestDeleteOrganizationRoles:
    @pytest.mark.usefixtures("_enable_organizations")
    def test_delete_role(self, db_request, monkeypatch):
        organization = OrganizationFactory.create(name="foobar")
        user = UserFactory.create(username="testuser")
        role = OrganizationRoleFactory.create(
            organization=organization,
            user=user,
            role_name=OrganizationRoleType.Owner,
        )
        user_2 = UserFactory.create()

        db_request.method = "POST"
        db_request.POST = MultiDict({"role_id": role.id})
        db_request.user = user_2
        db_request.session = pretend.stub(
            flash=pretend.call_recorder(lambda *a, **kw: None)
        )
        db_request.route_path = pretend.call_recorder(lambda *a, **kw: "/the-redirect")

        send_organization_member_removed_email = pretend.call_recorder(
            lambda *a, **kw: None
        )
        monkeypatch.setattr(
            org_views,
            "send_organization_member_removed_email",
            send_organization_member_removed_email,
        )
        send_removed_as_organization_member_email = pretend.call_recorder(
            lambda *a, **kw: None
        )
        monkeypatch.setattr(
            org_views,
            "send_removed_as_organization_member_email",
            send_removed_as_organization_member_email,
        )

        result = org_views.delete_organization_role(organization, db_request)

        assert db_request.route_path.calls == [
            pretend.call(
                "manage.organization.roles", organization_name=organization.name
            )
        ]
        assert db_request.db.query(OrganizationRole).all() == []
        assert send_organization_member_removed_email.calls == [
            pretend.call(
                db_request,
                set(),
                user=user,
                submitter=user_2,
                organization_name="foobar",
            )
        ]
        assert send_removed_as_organization_member_email.calls == [
            pretend.call(
                db_request,
                user,
                submitter=user_2,
                organization_name="foobar",
            )
        ]
        assert db_request.session.flash.calls == [
            pretend.call("Removed from organization", queue="success")
        ]
        assert isinstance(result, HTTPSeeOther)
        assert result.headers["Location"] == "/the-redirect"

    @pytest.mark.usefixtures("_enable_organizations")
    def test_delete_missing_role(self, db_request, monkeypatch):
        organization = OrganizationFactory.create(name="foobar")
        missing_role_id = str(uuid.uuid4())

        user_organizations = pretend.call_recorder(
            lambda *a, **kw: {
                "organizations_managed": [],
                "organizations_owned": [organization],
                "organizations_billing": [],
                "organizations_with_sole_owner": [],
            }
        )
        monkeypatch.setattr(org_views, "user_organizations", user_organizations)

        db_request.method = "POST"
        db_request.user = pretend.stub()
        db_request.POST = MultiDict({"role_id": missing_role_id})
        db_request.session = pretend.stub(
            flash=pretend.call_recorder(lambda *a, **kw: None)
        )
        db_request.route_path = pretend.call_recorder(lambda *a, **kw: "/the-redirect")

        result = org_views.delete_organization_role(organization, db_request)

        assert db_request.session.flash.calls == [
            pretend.call("Could not find member", queue="error")
        ]
        assert isinstance(result, HTTPSeeOther)
        assert result.headers["Location"] == "/the-redirect"

    @pytest.mark.usefixtures("_enable_organizations")
    def test_delete_other_role_as_nonowner(self, db_request):
        organization = OrganizationFactory.create(name="foobar")
        user = UserFactory.create(username="testuser")
        role = OrganizationRoleFactory.create(
            organization=organization,
            user=user,
            role_name=OrganizationRoleType.Owner,
        )
        user_2 = UserFactory.create()

        db_request.method = "POST"
        db_request.user = user_2
        db_request.POST = MultiDict({"role_id": role.id})
        db_request.has_permission = pretend.call_recorder(lambda *a, **kw: False)
        db_request.session = pretend.stub(
            flash=pretend.call_recorder(lambda *a, **kw: None)
        )
        db_request.route_path = pretend.call_recorder(lambda *a, **kw: "/the-redirect")

        result = org_views.delete_organization_role(organization, db_request)

        assert db_request.has_permission.calls == [
            pretend.call(Permissions.OrganizationsManage)
        ]
        assert db_request.session.flash.calls == [
            pretend.call(
                "Cannot remove other people from the organization", queue="error"
            )
        ]
        assert isinstance(result, HTTPSeeOther)
        assert result.headers["Location"] == "/the-redirect"

    @pytest.mark.usefixtures("_enable_organizations")
    def test_delete_own_owner_role(self, db_request):
        organization = OrganizationFactory.create(name="foobar")
        user = UserFactory.create(username="testuser")
        role = OrganizationRoleFactory.create(
            organization=organization,
            user=user,
            role_name=OrganizationRoleType.Owner,
        )

        db_request.method = "POST"
        db_request.user = user
        db_request.POST = MultiDict({"role_id": role.id})
        db_request.session = pretend.stub(
            flash=pretend.call_recorder(lambda *a, **kw: None)
        )
        db_request.route_path = pretend.call_recorder(lambda *a, **kw: "/the-redirect")

        result = org_views.delete_organization_role(organization, db_request)

        assert db_request.session.flash.calls == [
            pretend.call("Cannot remove yourself as Sole Owner", queue="error")
        ]
        assert isinstance(result, HTTPSeeOther)
        assert result.headers["Location"] == "/the-redirect"

    @pytest.mark.usefixtures("_enable_organizations")
    def test_delete_non_owner_role(self, db_request):
        organization = OrganizationFactory.create(name="foobar")
        user = UserFactory.create(username="testuser")
        role = OrganizationRoleFactory.create(
            organization=organization,
            user=user,
            role_name=OrganizationRoleType.Owner,
        )

        some_other_user = UserFactory.create(username="someotheruser")
        some_other_organization = OrganizationFactory.create(
            name="someotherorganization"
        )

        db_request.method = "POST"
        db_request.user = some_other_user
        db_request.POST = MultiDict({"role_id": role.id})
        db_request.session = pretend.stub(
            flash=pretend.call_recorder(lambda *a, **kw: None)
        )
        db_request.route_path = pretend.call_recorder(lambda *a, **kw: "/the-redirect")

        result = org_views.delete_organization_role(some_other_organization, db_request)

        assert db_request.session.flash.calls == [
            pretend.call("Could not find member", queue="error")
        ]
        assert isinstance(result, HTTPSeeOther)
        assert result.headers["Location"] == "/the-redirect"


class TestManageOrganizationHistory:
    def test_get(self, db_request, user_service):
        organization = OrganizationFactory.create()
        older_event = OrganizationEventFactory.create(
            source=organization,
            tag="fake:event",
            time=datetime.datetime(2017, 2, 5, 17, 18, 18, 462_634),
        )
        newer_event = OrganizationEventFactory.create(
            source=organization,
            tag="fake:event",
            time=datetime.datetime(2018, 2, 5, 17, 18, 18, 462_634),
        )

        assert org_views.manage_organization_history(organization, db_request) == {
            "events": [newer_event, older_event],
            "get_user": user_service.get_user,
            "organization": organization,
        }

    def test_raises_400_with_pagenum_type_str(self, monkeypatch, db_request):
        params = MultiDict({"page": "abc"})
        db_request.params = params

        events_query = pretend.stub()
        db_request.events_query = pretend.stub(
            events_query=lambda *a, **kw: events_query
        )

        page_obj = pretend.stub(page_count=10, item_count=1000)
        page_cls = pretend.call_recorder(lambda *a, **kw: page_obj)
        monkeypatch.setattr(views, "SQLAlchemyORMPage", page_cls)

        url_maker = pretend.stub()
        url_maker_factory = pretend.call_recorder(lambda request: url_maker)
        monkeypatch.setattr(views, "paginate_url_factory", url_maker_factory)

        organization = OrganizationFactory.create()
        with pytest.raises(HTTPBadRequest):
            org_views.manage_organization_history(organization, db_request)

        assert page_cls.calls == []

    def test_first_page(self, db_request, user_service):
        page_number = 1
        params = MultiDict({"page": page_number})
        db_request.params = params

        organization = OrganizationFactory.create()
        items_per_page = 25
        total_items = items_per_page + 2
        OrganizationEventFactory.create_batch(
            total_items, source=organization, tag="fake:event"
        )
        events_query = (
            db_request.db.query(Organization.Event)
            .join(Organization.Event.source)
            .filter(Organization.Event.source_id == organization.id)
            .order_by(Organization.Event.time.desc())
        )

        events_page = SQLAlchemyORMPage(
            events_query,
            page=page_number,
            items_per_page=items_per_page,
            item_count=total_items,
            url_maker=paginate_url_factory(db_request),
        )
        assert org_views.manage_organization_history(organization, db_request) == {
            "events": events_page,
            "get_user": user_service.get_user,
            "organization": organization,
        }

    def test_last_page(self, db_request, user_service):
        page_number = 2
        params = MultiDict({"page": page_number})
        db_request.params = params

        organization = OrganizationFactory.create()
        items_per_page = 25
        total_items = items_per_page + 2
        OrganizationEventFactory.create_batch(
            total_items, source=organization, tag="fake:event"
        )
        events_query = (
            db_request.db.query(Organization.Event)
            .join(Organization.Event.source)
            .filter(Organization.Event.source_id == organization.id)
            .order_by(Organization.Event.time.desc())
        )

        events_page = SQLAlchemyORMPage(
            events_query,
            page=page_number,
            items_per_page=items_per_page,
            item_count=total_items,
            url_maker=paginate_url_factory(db_request),
        )
        assert org_views.manage_organization_history(organization, db_request) == {
            "events": events_page,
            "get_user": user_service.get_user,
            "organization": organization,
        }

    def test_raises_404_with_out_of_range_page(self, db_request):
        page_number = 3
        params = MultiDict({"page": page_number})
        db_request.params = params

        organization = OrganizationFactory.create()
        items_per_page = 25
        total_items = items_per_page + 2
        OrganizationEventFactory.create_batch(
            total_items, source=organization, tag="fake:event"
        )

        with pytest.raises(HTTPNotFound):
            assert org_views.manage_organization_history(organization, db_request)


class TestManageOrganizationPublishingViews:
    def test_manage_organization_publishing_get(self, db_request):
        """Test GET request returns all forms and pending publishers"""
        organization = OrganizationFactory.create()
        user = UserFactory.create()
        db_request.POST = MultiDict()
        db_request.user = user
        db_request.route_url = pretend.call_recorder(lambda *a, **kw: "/fake/route")

        view = org_views.ManageOrganizationPublishingViews(organization, db_request)
        result = view.manage_organization_publishing()

        assert result["organization"] == organization
        assert "pending_github_publisher_form" in result
        assert "pending_gitlab_publisher_form" in result
        assert "pending_google_publisher_form" in result
        assert "pending_activestate_publisher_form" in result
        assert result["pending_oidc_publishers"] == organization.pending_oidc_publishers

    def test_add_pending_github_oidc_publisher_success(self, db_request, monkeypatch):
        """Test successfully adding a pending GitHub OIDC publisher"""
        organization = OrganizationFactory.create()
        user = UserFactory.create(with_verified_primary_email=True)
        db_request.POST = MultiDict()
        db_request.route_url = pretend.call_recorder(lambda *a, **kw: "/fake/route")
        db_request.session = pretend.stub(
            flash=pretend.call_recorder(lambda *a, **kw: None)
        )
        db_request.user = user

        # Mock form
        form = pretend.stub(
            validate=pretend.call_recorder(lambda: True),
            project_name=pretend.stub(data="test-project"),
            repository=pretend.stub(data="test-repo"),
            normalized_owner="test-owner",
            owner_id="12345",
            workflow_filename=pretend.stub(data="release.yml"),
            normalized_environment="",
        )
        monkeypatch.setattr(
            org_views, "PendingGitHubPublisherForm", lambda *a, **kw: form
        )

        view = org_views.ManageOrganizationPublishingViews(organization, db_request)
        result = view.add_pending_github_oidc_publisher()

        assert isinstance(result, HTTPSeeOther)
        assert db_request.session.flash.calls == [
            pretend.call(
                "Registered a new pending publisher to create the project "
                f"'test-project' owned by the '{organization.name}' organization.",
                queue="success",
            )
        ]
        assert db_request.metrics.increment.calls == [
            pretend.call(
                "warehouse.oidc.add_pending_publisher.attempt",
                tags=["publisher:GitHub", "organization:true"],
            ),
            pretend.call(
                "warehouse.oidc.add_pending_publisher.ok",
                tags=["publisher:GitHub", "organization:true"],
            ),
        ]

    def test_add_pending_github_oidc_publisher_disabled(self, db_request):
        """Test adding GitHub publisher when admin flag disables it"""
        organization = OrganizationFactory.create()
        user = UserFactory.create(with_verified_primary_email=True)
        db_request.user = user
        db_request.flags = pretend.stub(
            disallow_oidc=pretend.call_recorder(
                lambda flag: flag == AdminFlagValue.DISALLOW_GITHUB_OIDC
            )
        )
        db_request.session = pretend.stub(
            flash=pretend.call_recorder(lambda *a, **kw: None)
        )
        db_request.POST = MultiDict()
        db_request.route_url = pretend.call_recorder(lambda *a, **kw: "/fake/route")

        view = org_views.ManageOrganizationPublishingViews(organization, db_request)
        result = view.add_pending_github_oidc_publisher()

        assert result == view.default_response
        assert db_request.session.flash.calls == [
            pretend.call(
                "GitHub-based trusted publishing is temporarily disabled. "
                "See https://pypi.org/help#admin-intervention for details.",
                queue="error",
            )
        ]

    def test_add_pending_github_oidc_publisher_over_limit(self, db_request):
        """Adding GitHub publisher fails when org has too many pending publishers"""
        organization = OrganizationFactory.create()
        user = UserFactory.create(with_verified_primary_email=True)
        # Add 3 existing pending publishers
        PendingGitHubPublisherFactory.create_batch(3, organization_id=organization.id)

        db_request.user = user
        db_request.session = pretend.stub(
            flash=pretend.call_recorder(lambda *a, **kw: None)
        )
        db_request.POST = MultiDict()
        db_request.route_url = pretend.call_recorder(lambda *a, **kw: "/fake/route")

        view = org_views.ManageOrganizationPublishingViews(organization, db_request)
        result = view.add_pending_github_oidc_publisher()

        assert result == view.default_response
        assert db_request.session.flash.calls == [
            pretend.call(
                "The trusted publisher could not be registered",
                queue="error",
            )
        ]

    def test_add_pending_gitlab_oidc_publisher_success(self, db_request, monkeypatch):
        """Test successfully adding a pending GitLab OIDC publisher"""
        organization = OrganizationFactory.create()
        user = UserFactory.create(with_verified_primary_email=True)
        db_request.flags = pretend.stub(
            disallow_oidc=pretend.call_recorder(lambda *a: False)
        )
        db_request.POST = MultiDict()
        db_request.path = "/fake/path"
        db_request.route_url = pretend.call_recorder(lambda *a, **kw: "/fake/route")
        db_request.user = user

        # Mock form
        form = pretend.stub(
            validate=pretend.call_recorder(lambda: True),
            project_name=pretend.stub(data="test-project"),
            namespace=pretend.stub(data="test-namespace"),
            project=pretend.stub(data="test-project"),
            workflow_filepath=pretend.stub(data=".gitlab-ci.yml"),
            environment=pretend.stub(data=""),
            issuer_url=pretend.stub(data="https://gitlab.com"),
        )
        monkeypatch.setattr(
            org_views, "PendingGitLabPublisherForm", lambda *a, **kw: form
        )

        view = org_views.ManageOrganizationPublishingViews(organization, db_request)
        result = view.add_pending_gitlab_oidc_publisher()

        assert isinstance(result, HTTPSeeOther)
        assert db_request.metrics.increment.calls[-1] == pretend.call(
            "warehouse.oidc.add_pending_publisher.ok",
            tags=["publisher:GitLab", "organization:true"],
        )

    def test_gitlab_form_includes_issuer_url_choices(self, db_request, monkeypatch):
        """Test that GitLab form is created with issuer_url_choices"""
        organization = OrganizationFactory.create()
        user = UserFactory.create()
        db_request.POST = MultiDict()
        db_request.user = user
        db_request.route_url = pretend.call_recorder(lambda *a, **kw: "/fake/route")

        # Mock GitLabPublisher.get_available_issuer_urls to return multiple issuers
        mock_issuers = [
            ("https://gitlab.com", "GitLab.com"),
            ("https://gitlab.example.com", "Custom GitLab"),
        ]
        monkeypatch.setattr(
            org_views.GitLabPublisher,
            "get_available_issuer_urls",
            lambda organization: mock_issuers,
        )

        # Track the form creation to verify issuer_url_choices are passed
        form_calls = []

        def track_form_creation(*args, **kwargs):
            form_calls.append(kwargs)
            return pretend.stub(
                validate=lambda: False,
                project_name=pretend.stub(data=""),
                namespace=pretend.stub(data=""),
                project=pretend.stub(data=""),
                workflow_filepath=pretend.stub(data=""),
                environment=pretend.stub(data=""),
                issuer_url=pretend.stub(data="", choices=mock_issuers),
            )

        monkeypatch.setattr(
            org_views, "PendingGitLabPublisherForm", track_form_creation
        )

        view = org_views.ManageOrganizationPublishingViews(organization, db_request)

        # Verify that the form was created with issuer_url_choices
        assert len(form_calls) == 1
        assert form_calls[0]["issuer_url_choices"] == mock_issuers

        # Verify the form has the correct choices
        assert view.pending_gitlab_publisher_form.issuer_url.choices == mock_issuers

    def test_manage_organization_publishing_get_oidc_disabled(
        self, db_request, monkeypatch
    ):
        """Test GET request when global OIDC is disabled"""
        organization = OrganizationFactory.create()
        user = UserFactory.create()
        db_request.POST = MultiDict()
        db_request.user = user
        db_request.route_url = pretend.call_recorder(lambda *a, **kw: "/fake/route")
        db_request.flags = pretend.stub(
            disallow_oidc=pretend.call_recorder(lambda f=None: True)
        )
        db_request.session = pretend.stub(
            flash=pretend.call_recorder(lambda *a, **kw: None)
        )

        # Mock all form classes since default_response tries to instantiate them
        pending_github_publisher_form_obj = pretend.stub()
        monkeypatch.setattr(
            org_views,
            "PendingGitHubPublisherForm",
            lambda *a, **kw: pending_github_publisher_form_obj,
        )
        pending_gitlab_publisher_form_obj = pretend.stub()
        monkeypatch.setattr(
            org_views,
            "PendingGitLabPublisherForm",
            lambda *a, **kw: pending_gitlab_publisher_form_obj,
        )
        pending_google_publisher_form_obj = pretend.stub()
        monkeypatch.setattr(
            org_views,
            "PendingGooglePublisherForm",
            lambda *a, **kw: pending_google_publisher_form_obj,
        )
        pending_activestate_publisher_form_obj = pretend.stub()
        monkeypatch.setattr(
            org_views,
            "PendingActiveStatePublisherForm",
            lambda *a, **kw: pending_activestate_publisher_form_obj,
        )

        view = org_views.ManageOrganizationPublishingViews(organization, db_request)
        result = view.manage_organization_publishing()

        assert result["organization"] == organization
        assert db_request.session.flash.calls == [
            pretend.call(
                "Trusted publishing is temporarily disabled. "
                "See https://pypi.org/help#admin-intervention for details.",
                queue="error",
            )
        ]

    def test_add_pending_github_oidc_publisher_already_exists(
        self, db_request, monkeypatch
    ):
        """Test adding GitHub publisher when it already exists"""
        organization = OrganizationFactory.create()
        user = UserFactory.create(with_verified_primary_email=True)

        # Create an existing pending publisher with matching attributes
        existing_publisher = PendingGitHubPublisherFactory.create(
            project_name="test-project",
            repository_name="test-repo",
            repository_owner="test-owner",
            repository_owner_id="12345",
            workflow_filename="release.yml",
            environment="",
            organization_id=organization.id,
        )
        db_request.db.add(existing_publisher)
        db_request.db.flush()

        db_request.POST = MultiDict()
        db_request.route_url = pretend.call_recorder(lambda *a, **kw: "/fake/route")
        db_request.session = pretend.stub(
            flash=pretend.call_recorder(lambda *a, **kw: None)
        )
        db_request.user = user

        # Mock form with same data as existing publisher
        form = pretend.stub(
            validate=pretend.call_recorder(lambda: True),
            project_name=pretend.stub(data="test-project"),
            repository=pretend.stub(data="test-repo"),
            normalized_owner="test-owner",
            owner_id="12345",
            workflow_filename=pretend.stub(data="release.yml"),
            normalized_environment="",
        )
        monkeypatch.setattr(
            org_views, "PendingGitHubPublisherForm", lambda *a, **kw: form
        )

        view = org_views.ManageOrganizationPublishingViews(organization, db_request)
        result = view.add_pending_github_oidc_publisher()

        assert result == view.default_response
        assert db_request.session.flash.calls == [
            pretend.call(
                "This publisher has already been registered in your organization. "
                "See your existing pending publishers below.",
                queue="error",
            )
        ]

    def test_add_pending_github_oidc_publisher_unique_violation(
        self, db_request, monkeypatch
    ):
        """Test UniqueViolation exception handling during publisher creation"""
        organization = OrganizationFactory.create()
        user = UserFactory.create(with_verified_primary_email=True)
        db_request.POST = MultiDict()
        db_request.path = "/fake/path"
        db_request.route_url = pretend.call_recorder(lambda *a, **kw: "/fake/route")
        db_request.user = user

        # Mock db.add to raise UniqueViolation (simulates race condition)
        db_request.db.add = pretend.raiser(UniqueViolation("foo", "bar", "baz"))

        # Mock form
        form = pretend.stub(
            validate=pretend.call_recorder(lambda: True),
            project_name=pretend.stub(data="test-project"),
            repository=pretend.stub(data="test-repo"),
            normalized_owner="test-owner",
            owner_id="12345",
            workflow_filename=pretend.stub(data="release.yml"),
            normalized_environment="",
        )
        monkeypatch.setattr(
            org_views, "PendingGitHubPublisherForm", lambda *a, **kw: form
        )

        view = org_views.ManageOrganizationPublishingViews(organization, db_request)
        result = view.add_pending_github_oidc_publisher()

        # Should return HTTPSeeOther redirect (double-post protection)
        assert isinstance(result, HTTPSeeOther)
        assert result.location == "/fake/path"

    def test_two_orgs_can_create_pending_publishers_for_same_project_name(
        self, db_request, monkeypatch
    ):
        """
        Two different organizations can create pending OIDC publishers
        for the same future project name but with different OIDC credentials.
        """
        # Create two separate organizations with different owners
        org1_owner = UserFactory.create(username="org1-owner")
        org2_owner = UserFactory.create(username="org2-owner")
        EmailFactory.create(user=org1_owner, verified=True, primary=True)
        EmailFactory.create(user=org2_owner, verified=True, primary=True)

        org1 = OrganizationFactory.create(name="org1")
        org2 = OrganizationFactory.create(name="org2")

        OrganizationRoleFactory.create(
            organization=org1, user=org1_owner, role_name="Owner"
        )
        OrganizationRoleFactory.create(
            organization=org2, user=org2_owner, role_name="Owner"
        )

        # Setup request for org1
        db_request.user = org1_owner
        db_request.flags = pretend.stub(
            disallow_oidc=pretend.call_recorder(lambda f=None: False)
        )
        db_request.session = pretend.stub(
            flash=pretend.call_recorder(lambda *a, **kw: None)
        )
        db_request.registry = pretend.stub(settings={"github.token": "fake-api-token"})
        db_request.POST = MultiDict(
            {
                "project_name": "same-project-name",
                "owner": "org1",
                "repository": "repo1",
                "workflow_filename": "release.yml",
                "environment": "",
            }
        )

        # Mock the GitHub API lookup for org1
        monkeypatch.setattr(
            org_views.PendingGitHubPublisherForm,
            "_lookup_owner",
            lambda *a: {"login": "org1", "id": "11111"},
        )

        # Org1 creates a pending publisher
        view1 = org_views.ManageOrganizationPublishingViews(org1, db_request)
        result1 = view1.add_pending_github_oidc_publisher()
        assert isinstance(result1, HTTPSeeOther)

        # Verify org1's pending publisher was created
        pending_publisher_1 = (
            db_request.db.query(PendingGitHubPublisher)
            .filter(
                PendingGitHubPublisher.organization_id == org1.id,
                PendingGitHubPublisher.project_name == "same-project-name",
            )
            .one()
        )
        assert pending_publisher_1.repository_owner == "org1"
        assert pending_publisher_1.repository_name == "repo1"
        assert pending_publisher_1.organization_id == org1.id

        # Setup request for org2 with SAME project name but different credentials
        db_request.user = org2_owner
        db_request.POST = MultiDict(
            {
                "project_name": "same-project-name",  # SAME PROJECT NAME
                "owner": "org2",  # DIFFERENT GitHub org
                "repository": "repo2",  # DIFFERENT repo
                "workflow_filename": "release.yml",
                "environment": "",
            }
        )

        # Mock the GitHub API lookup for org2
        monkeypatch.setattr(
            org_views.PendingGitHubPublisherForm,
            "_lookup_owner",
            lambda *a: {"login": "org2", "id": "22222"},
        )

        # Org2 creates a pending publisher for the SAME project name
        view2 = org_views.ManageOrganizationPublishingViews(org2, db_request)
        result2 = view2.add_pending_github_oidc_publisher()
        assert isinstance(result2, HTTPSeeOther)

        # Verify org2's pending publisher was also created
        pending_publisher_2 = (
            db_request.db.query(PendingGitHubPublisher)
            .filter(
                PendingGitHubPublisher.organization_id == org2.id,
                PendingGitHubPublisher.project_name == "same-project-name",
            )
            .one()
        )
        assert pending_publisher_2.repository_owner == "org2"
        assert pending_publisher_2.repository_name == "repo2"
        assert pending_publisher_2.organization_id == org2.id

        # CRITICAL: Both pending publishers should exist
        all_pending_publishers = (
            db_request.db.query(PendingGitHubPublisher)
            .filter(PendingGitHubPublisher.project_name == "same-project-name")
            .all()
        )
        assert len(all_pending_publishers) == 2

        # Verify they have different credentials (different organizations)
        assert pending_publisher_1.id != pending_publisher_2.id
        assert (
            pending_publisher_1.repository_owner != pending_publisher_2.repository_owner
        )
        assert (
            pending_publisher_1.repository_name != pending_publisher_2.repository_name
        )
